fix: remove metric units that are interpreted as ratios (#49)

Metrics with a unit of "1" are interpreted as ratios by the prometheus
exporter, which has the side effect of appending `_ratio` to the metric
name. This is supposed to only be done for gauges but is actually
applied to counters too. See
https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/pkg/translator/prometheus
for the transformations applied by the prometheus exporter. Since we
were using "1" for dimensionless counts I have simply removed the units.

This PR also hooks up the SentMessages, SentMessageErrors, SentRequests,
SentRequestErrors, OutboundRequestLatency and SentBytes DHT metrics.

Author: iand

dht.go

@@ -132,7 +132,8 @@ func New(h host.Host, cfg *Config) (*DHT, error) {
 	rtr := &router{
 		host:       h,
 		protocolID: cfg.ProtocolID,
-		tracer:     d.tele.Tracer,
+		tele:       d.tele,
+		clk:        cfg.Clock,
 	}
 	d.kad, err = coord.NewCoordinator(kadt.PeerID(d.host.ID()), rtr, d.rt, coordCfg)
 	if err != nil {

internal/coord/routing/bootstrap.go

@@ -138,7 +138,6 @@ func NewBootstrap[K kad.Key[K], N kad.NodeID[K]](self N, cfg *BootstrapConfig) (
 	b.counterFindSent, err = cfg.Meter.Int64Counter(
 		"bootstrap_find_sent",
 		metric.WithDescription("Total number of find closer nodes requests sent by the bootstrap state machine"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create bootstrap_find_sent counter: %w", err)
@@ -147,7 +146,6 @@ func NewBootstrap[K kad.Key[K], N kad.NodeID[K]](self N, cfg *BootstrapConfig) (
 	b.counterFindSucceeded, err = cfg.Meter.Int64Counter(
 		"bootstrap_find_succeeded",
 		metric.WithDescription("Total number of find closer nodes requests sent by the bootstrap state machine that were successful"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create bootstrap_find_succeeded counter: %w", err)
@@ -156,7 +154,6 @@ func NewBootstrap[K kad.Key[K], N kad.NodeID[K]](self N, cfg *BootstrapConfig) (
 	b.counterFindFailed, err = cfg.Meter.Int64Counter(
 		"bootstrap_find_failed",
 		metric.WithDescription("Total number of find closer nodes requests sent by the bootstrap state machine that failed"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create bootstrap_find_failed counter: %w", err)
@@ -165,7 +162,6 @@ func NewBootstrap[K kad.Key[K], N kad.NodeID[K]](self N, cfg *BootstrapConfig) (
 	b.gaugeRunning, err = cfg.Meter.Int64ObservableGauge(
 		"bootstrap_running",
 		metric.WithDescription("Whether or not the bootstrap is running"),
-		metric.WithUnit("1"),
 		metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
 			if b.running.Load() {
 				o.Observe(1)

internal/coord/routing/explore.go

@@ -186,7 +186,6 @@ func NewExplore[K kad.Key[K], N kad.NodeID[K]](self N, rt RoutingTableCpl[K, N],
 	e.counterFindSent, err = cfg.Meter.Int64Counter(
 		"explore_find_sent",
 		metric.WithDescription("Total number of find closer nodes requests sent by the explore state machine"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create explore_find_sent counter: %w", err)
@@ -195,7 +194,6 @@ func NewExplore[K kad.Key[K], N kad.NodeID[K]](self N, rt RoutingTableCpl[K, N],
 	e.counterFindSucceeded, err = cfg.Meter.Int64Counter(
 		"explore_find_succeeded",
 		metric.WithDescription("Total number of find closer nodes requests sent by the explore state machine that were successful"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create explore_find_succeeded counter: %w", err)
@@ -204,7 +202,6 @@ func NewExplore[K kad.Key[K], N kad.NodeID[K]](self N, rt RoutingTableCpl[K, N],
 	e.counterFindFailed, err = cfg.Meter.Int64Counter(
 		"explore_find_failed",
 		metric.WithDescription("Total number of find closer nodes requests sent by the explore state machine that failed"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create explore_find_failed counter: %w", err)
@@ -213,7 +210,6 @@ func NewExplore[K kad.Key[K], N kad.NodeID[K]](self N, rt RoutingTableCpl[K, N],
 	e.gaugeRunning, err = cfg.Meter.Int64ObservableGauge(
 		"explore_running",
 		metric.WithDescription("Whether or not the an explore is running for a cpl"),
-		metric.WithUnit("1"),
 		metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
 			if e.running.Load() {
 				o.Observe(1, metric.WithAttributeSet(e.cplAttributeSet.Load().(attribute.Set)))

internal/coord/routing/include.go

@@ -149,7 +149,6 @@ func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *I
 	in.counterChecksSent, err = cfg.Meter.Int64Counter(
 		"include_checks_sent",
 		metric.WithDescription("Total number of connectivity checks sent by the include state machine"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_sent counter: %w", err)
@@ -158,7 +157,6 @@ func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *I
 	in.counterChecksPassed, err = cfg.Meter.Int64Counter(
 		"include_checks_passed",
 		metric.WithDescription("Total number of connectivity checks sent by the include state machine that were successful"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_passed counter: %w", err)
@@ -167,7 +165,6 @@ func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *I
 	in.counterChecksFailed, err = cfg.Meter.Int64Counter(
 		"include_checks_failed",
 		metric.WithDescription("Total number of connectivity checks sent by the include state machine that failed"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_failed counter: %w", err)
@@ -176,7 +173,6 @@ func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *I
 	in.counterCandidatesDroppedCapacity, err = cfg.Meter.Int64Counter(
 		"include_candidates_dropped_capacity",
 		metric.WithDescription("Total number of nodes that were not added to the candidate queue because it was already at maximum capacity"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_candidates_dropped_capacity counter: %w", err)
@@ -185,7 +181,6 @@ func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *I
 	in.gaugeCandidateCount, err = cfg.Meter.Int64ObservableGauge(
 		"include_candidate_count",
 		metric.WithDescription("Total number of nodes in the include state machine's candidate queue"),
-		metric.WithUnit("1"),
 		metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
 			o.Observe(in.candidateCount.Load())
 			return nil

internal/coord/routing/probe.go

@@ -177,7 +177,6 @@ func NewProbe[K kad.Key[K], N kad.NodeID[K]](rt RoutingTableCpl[K, N], cfg *Prob
 	p.counterChecksSent, err = cfg.Meter.Int64Counter(
 		"probe_checks_sent",
 		metric.WithDescription("Total number of connectivity checks sent by the probe state machine"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create probe_checks_sent counter: %w", err)
@@ -186,7 +185,6 @@ func NewProbe[K kad.Key[K], N kad.NodeID[K]](rt RoutingTableCpl[K, N], cfg *Prob
 	p.counterChecksPassed, err = cfg.Meter.Int64Counter(
 		"probe_checks_passed",
 		metric.WithDescription("Total number of connectivity checks sent by the probe state machine that were successful"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create probe_checks_passed counter: %w", err)
@@ -195,7 +193,6 @@ func NewProbe[K kad.Key[K], N kad.NodeID[K]](rt RoutingTableCpl[K, N], cfg *Prob
 	p.counterChecksFailed, err = cfg.Meter.Int64Counter(
 		"probe_checks_failed",
 		metric.WithDescription("Total number of connectivity checks sent by the probe state machine that failed"),
-		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create probe_checks_failed counter: %w", err)
@@ -204,7 +201,6 @@ func NewProbe[K kad.Key[K], N kad.NodeID[K]](rt RoutingTableCpl[K, N], cfg *Prob
 	p.gaugePendingCount, err = cfg.Meter.Int64ObservableGauge(
 		"probe_pending_count",
 		metric.WithDescription("Total number of nodes being monitored by the probe state machine"),
-		metric.WithUnit("1"),
 		metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
 			o.Observe(p.pendingCount.Load())
 			return nil

pb/msg.aux.go

@@ -3,6 +3,7 @@ package pb
 import (
 	"bytes"
 	"fmt"
+	math_bits "math/bits"
 
 	"github.com/libp2p/go-libp2p/core/peer"
 	ma "github.com/multiformats/go-multiaddr"
@@ -114,6 +115,41 @@ func (m *Message) CloserNodes() []kadt.PeerID {
 	return ids
 }
 
+func (m *Message) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Type != 0 {
+		n += 1 + sovDht(uint64(m.Type))
+	}
+	l = len(m.Key)
+	if l > 0 {
+		n += 1 + l + sovDht(uint64(l))
+	}
+	if m.Record != nil {
+		l = m.Record.Size()
+		n += 1 + l + sovDht(uint64(l))
+	}
+	if len(m.CloserPeers) > 0 {
+		for _, e := range m.CloserPeers {
+			l = e.Size()
+			n += 1 + l + sovDht(uint64(l))
+		}
+	}
+	if len(m.ProviderPeers) > 0 {
+		for _, e := range m.ProviderPeers {
+			l = e.Size()
+			n += 1 + l + sovDht(uint64(l))
+		}
+	}
+	if m.ClusterLevelRaw != 0 {
+		n += 1 + sovDht(uint64(m.ClusterLevelRaw))
+	}
+	return n
+}
+
 // Addresses returns the Multiaddresses associated with the Message_Peer entry
 func (m *Message_Peer) Addresses() []ma.Multiaddr {
 	if m == nil {
@@ -133,3 +169,27 @@ func (m *Message_Peer) Addresses() []ma.Multiaddr {
 
 	return maddrs
 }
+
+func (m *Message_Peer) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Id)
+	n += 1 + l + sovDht(uint64(l))
+	if len(m.Addrs) > 0 {
+		for _, b := range m.Addrs {
+			l = len(b)
+			n += 1 + l + sovDht(uint64(l))
+		}
+	}
+	if m.Connection != 0 {
+		n += 1 + sovDht(uint64(m.Connection))
+	}
+	return n
+}
+
+func sovDht(x uint64) (n int) {
+	return (math_bits.Len64(x|1) + 6) / 7
+}

router.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/benbjohnson/clock"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -31,8 +32,10 @@ type router struct {
 	// [protocol]: https://docs.libp2p.io/concepts/fundamentals/protocols/
 	protocolID protocol.ID
 
-	// an open telemetry tacer instance
-	tracer trace.Tracer
+	// tele holds a reference to a telemetry struct
+	tele *Telemetry
+
+	clk clock.Clock
 }
 
 var _ coordt.Router[kadt.Key, kadt.PeerID, *pb.Message] = (*router)(nil)
@@ -43,7 +46,7 @@ func (r *router) SendMessage(ctx context.Context, to kadt.PeerID, req *pb.Messag
 		trace.WithAttributes(tele.AttrPeerID(to.String())),
 		trace.WithAttributes(tele.AttrKey(base64.RawStdEncoding.EncodeToString(req.GetKey()))),
 	}
-	ctx, span := r.tracer.Start(ctx, "router.SendMessage", spanOpts...)
+	ctx, span := r.tele.Tracer.Start(ctx, "router.SendMessage", spanOpts...)
 	defer func() {
 		if err != nil {
 			span.RecordError(err)
@@ -70,27 +73,42 @@ func (r *router) SendMessage(ctx context.Context, to kadt.PeerID, req *pb.Messag
 	w := pbio.NewDelimitedWriter(s)
 	reader := msgio.NewVarintReaderSize(s, network.MessageSizeMax)
 
+	if !req.ExpectResponse() {
+		err = w.WriteMsg(req)
+		r.tele.SentMessages.Add(ctx, 1)
+		if err != nil {
+			r.tele.SentMessageErrors.Add(ctx, 1)
+			return nil, fmt.Errorf("write message: %w", err)
+		}
+		r.tele.SentBytes.Record(ctx, int64(req.Size()))
+		return nil, nil
+	}
+
+	start := r.clk.Now()
+
 	err = w.WriteMsg(req)
+	r.tele.SentRequests.Add(ctx, 1)
 	if err != nil {
+		r.tele.SentRequestErrors.Add(ctx, 1)
 		return nil, fmt.Errorf("write message: %w", err)
 	}
-
-	if !req.ExpectResponse() {
-		return nil, nil
-	}
+	r.tele.SentBytes.Record(ctx, int64(req.Size()))
 
 	span.End()
-	ctx, span = r.tracer.Start(ctx, "router.ReadMessage", spanOpts...)
+	ctx, span = r.tele.Tracer.Start(ctx, "router.ReadMessage", spanOpts...)
 
 	data, err := reader.ReadMsg()
 	if err != nil {
+		r.tele.SentRequestErrors.Add(ctx, 1)
 		return nil, fmt.Errorf("read message: %w", err)
 	}
 
 	protoResp := pb.Message{}
 	if err = proto.Unmarshal(data, &protoResp); err != nil {
+		r.tele.SentRequestErrors.Add(ctx, 1)
 		return nil, err
 	}
+	r.tele.OutboundRequestLatency.Record(ctx, float64(r.clk.Since(start))/float64(time.Millisecond))
 
 	for _, info := range protoResp.CloserPeersAddrInfos() {
 		_ = r.addToPeerStore(ctx, info, time.Hour) // TODO: replace hard coded time.Hour with config

telemetry.go

@@ -24,9 +24,9 @@ type Telemetry struct {
 	ReceivedBytes          metric.Int64Histogram
 	InboundRequestLatency  metric.Float64Histogram
 	OutboundRequestLatency metric.Float64Histogram
-	SentMessages           metric.Int64Counter
+	SentMessages           metric.Int64Counter // number of messages sent that did not expect a response
 	SentMessageErrors      metric.Int64Counter
-	SentRequests           metric.Int64Counter
+	SentRequests           metric.Int64Counter // number of messages sent that expected a response
 	SentRequestErrors      metric.Int64Counter
 	SentBytes              metric.Int64Histogram
 	LRUCache               metric.Int64Counter
@@ -63,12 +63,12 @@ func NewTelemetry(meterProvider metric.MeterProvider, tracerProvider trace.Trace
 
 	// Initalize metrics for the DHT
 
-	t.ReceivedMessages, err = meter.Int64Counter("received_messages", metric.WithDescription("Total number of messages received per RPC"), metric.WithUnit("1"))
+	t.ReceivedMessages, err = meter.Int64Counter("received_messages", metric.WithDescription("Total number of messages received per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("received_messages counter: %w", err)
 	}
 
-	t.ReceivedMessageErrors, err = meter.Int64Counter("received_message_errors", metric.WithDescription("Total number of errors for messages received per RPC"), metric.WithUnit("1"))
+	t.ReceivedMessageErrors, err = meter.Int64Counter("received_message_errors", metric.WithDescription("Total number of errors for messages received per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("received_message_errors counter: %w", err)
 	}
@@ -88,22 +88,22 @@ func NewTelemetry(meterProvider metric.MeterProvider, tracerProvider trace.Trace
 		return nil, fmt.Errorf("outbound_request_latency histogram: %w", err)
 	}
 
-	t.SentMessages, err = meter.Int64Counter("sent_messages", metric.WithDescription("Total number of messages sent per RPC"), metric.WithUnit("1"))
+	t.SentMessages, err = meter.Int64Counter("sent_messages", metric.WithDescription("Total number of messages sent per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("sent_messages counter: %w", err)
 	}
 
-	t.SentMessageErrors, err = meter.Int64Counter("sent_message_errors", metric.WithDescription("Total number of errors for messages sent per RPC"), metric.WithUnit("1"))
+	t.SentMessageErrors, err = meter.Int64Counter("sent_message_errors", metric.WithDescription("Total number of errors for messages sent per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("sent_message_errors counter: %w", err)
 	}
 
-	t.SentRequests, err = meter.Int64Counter("sent_requests", metric.WithDescription("Total number of requests sent per RPC"), metric.WithUnit("1"))
+	t.SentRequests, err = meter.Int64Counter("sent_requests", metric.WithDescription("Total number of requests sent per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("sent_requests counter: %w", err)
 	}
 
-	t.SentRequestErrors, err = meter.Int64Counter("sent_request_errors", metric.WithDescription("Total number of errors for requests sent per RPC"), metric.WithUnit("1"))
+	t.SentRequestErrors, err = meter.Int64Counter("sent_request_errors", metric.WithDescription("Total number of errors for requests sent per RPC"))
 	if err != nil {
 		return nil, fmt.Errorf("sent_request_errors counter: %w", err)
 	}
@@ -113,12 +113,12 @@ func NewTelemetry(meterProvider metric.MeterProvider, tracerProvider trace.Trace
 		return nil, fmt.Errorf("sent_bytes histogram: %w", err)
 	}
 
-	t.LRUCache, err = meter.Int64Counter("lru_cache", metric.WithDescription("Cache hit or miss counter"), metric.WithUnit("1"))
+	t.LRUCache, err = meter.Int64Counter("lru_cache", metric.WithDescription("Cache hit or miss counter"))
 	if err != nil {
 		return nil, fmt.Errorf("lru_cache counter: %w", err)
 	}
 
-	t.NetworkSize, err = meter.Int64Counter("network_size", metric.WithDescription("Network size estimation"), metric.WithUnit("1"))
+	t.NetworkSize, err = meter.Int64Counter("network_size", metric.WithDescription("Network size estimation"))
 	if err != nil {
 		return nil, fmt.Errorf("network_size counter: %w", err)
 	}