feat(telemetry): add blackbox instrumentation to the header module + share module + p2p bandwidth metrics

.gitignore

@@ -20,3 +20,4 @@ vendor
 /cel-key
 coverage.txt
 go.work
+nodebuilder/keyring-test/

core/testing_grpc.go

@@ -109,6 +109,5 @@ func startGRPCServer(
 			log.Error("serving GRPC: ", err)
 		}
 	}()
-
 	return grpcSrv, nil
 }

das/metrics.go

@@ -141,9 +141,11 @@ func (m *metrics) observeSample(
 	if m == nil {
 		return
 	}
+
 	m.sampleTime.Record(ctx, sampleTime.Seconds(),
 		attribute.Bool("failed", err != nil),
 		attribute.Int("header_width", len(h.DAH.RowsRoots)),
+		attribute.Int("header", int(h.RawHeader.Height)),
 	)
 
 	m.sampled.Add(ctx, 1,

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/celestiaorg/go-libp2p-messenger v0.1.0
 	github.com/celestiaorg/nmt v0.14.0
 	github.com/celestiaorg/rsmt2d v0.8.0
+	github.com/celestiaorg/utils v0.1.0
 	github.com/cosmos/cosmos-sdk v0.46.7
 	github.com/cosmos/cosmos-sdk/api v0.1.0
 	github.com/cristalhq/jwt v1.2.0

go.sum

@@ -221,6 +221,8 @@ github.com/celestiaorg/quantum-gravity-bridge v1.3.0 h1:9zPIp7w1FWfkPnn16y3S4FpF
 github.com/celestiaorg/quantum-gravity-bridge v1.3.0/go.mod h1:6WOajINTDEUXpSj5UZzod16UZ96ZVB/rFNKyM+Mt1gI=
 github.com/celestiaorg/rsmt2d v0.8.0 h1:ZUxTCELZCM9zMGKNF3cT+rUqMddXMeiuyleSJPZ3Wn4=
 github.com/celestiaorg/rsmt2d v0.8.0/go.mod h1:hhlsTi6G3+X5jOP/8Lb/d7i5y2XNFmnyMddYbFSmrgo=
+github.com/celestiaorg/utils v0.1.0 h1:WsP3O8jF7jKRgLNFmlDCwdThwOFMFxg0MnqhkLFVxPo=
+github.com/celestiaorg/utils v0.1.0/go.mod h1:vQTh7MHnvpIeCQZ2/Ph+w7K1R2UerDheZbgJEJD2hSU=
 github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=

libs/header/p2p/exchange.go

@@ -33,6 +33,8 @@ type Exchange[H header.Header] struct {
 	peerTracker  *peerTracker
 
 	Params ClientParameters
+
+	metrics *metrics
 }
 
 func NewExchange[H header.Header](
@@ -251,7 +253,9 @@ func (ex *Exchange[H]) request(
 	req *p2p_pb.HeaderRequest,
 ) ([]H, error) {
 	log.Debugw("requesting peer", "peer", to)
-	responses, _, _, err := sendMessage(ctx, ex.host, to, ex.protocolID, req)
+	responses, size, duration, err := sendMessage(ctx, ex.host, to, ex.protocolID, req)
+	ex.metrics.ObserveRequest(ctx, size, duration)
+
 	if err != nil {
 		log.Debugw("err sending request", "peer", to, "err", err)
 		return nil, err

libs/header/p2p/metrics.go

@@ -0,0 +1,57 @@
+package p2p
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/metric/global"
+	"go.opentelemetry.io/otel/metric/instrument"
+	"go.opentelemetry.io/otel/metric/instrument/syncfloat64"
+)
+
+type metrics struct {
+	responseSize     syncfloat64.Histogram
+	responseDuration syncfloat64.Histogram
+}
+
+var (
+	meter = global.MeterProvider().Meter("libs/header/p2p")
+)
+
+func (ex *Exchange[H]) RegisterMetrics() error {
+	responseSize, err := meter.
+		SyncFloat64().
+		Histogram(
+			"libhead_get_headers_response_size",
+			instrument.WithDescription("Size of get headers response in bytes"),
+		)
+	if err != nil {
+		panic(err)
+	}
+
+	responseDuration, err := meter.
+		SyncFloat64().
+		Histogram(
+			"libhead_get_headers_request_duration",
+			instrument.WithDescription("Duration of get headers request in seconds"),
+		)
+	if err != nil {
+		return err
+	}
+
+	m := &metrics{
+		responseSize:     responseSize,
+		responseDuration: responseDuration,
+	}
+
+	ex.metrics = m
+
+	return nil
+}
+
+func (m *metrics) ObserveRequest(ctx context.Context, size uint64, duration uint64) {
+	if m == nil {
+		return
+	}
+	m.responseSize.Record(ctx, float64(size))
+	m.responseDuration.Record(ctx, float64(duration))
+}

nodebuilder/header/header.go

@@ -18,6 +18,9 @@ type Module interface {
 	Head(context.Context) (*header.ExtendedHeader, error)
 	// IsSyncing returns the status of sync
 	IsSyncing(context.Context) bool
+
+	// SyncerHead returns the highest known ExtendedHeader of the network.
+	SyncerHead(context.Context) (*header.ExtendedHeader, error)
 }
 
 // API is a wrapper around Module for the RPC.

nodebuilder/header/opts.go

@@ -0,0 +1,20 @@
+package header
+
+import (
+	header "github.com/celestiaorg/celestia-node/header"
+	libhead "github.com/celestiaorg/celestia-node/libs/header"
+	p2p "github.com/celestiaorg/celestia-node/libs/header/p2p"
+)
+
+// WithMetrics provides sets `MetricsEnabled` to true on ClientParameters for the header exchange
+func WithMetrics(ex libhead.Exchange[*header.ExtendedHeader]) error {
+	exchange, ok := (ex).(*p2p.Exchange[*header.ExtendedHeader])
+	if !ok {
+		// not all implementations of libhead.Exchange[*header.ExtendedHeader]
+		// are p2p.Exchange[*header.ExtendedHeader
+		// thus we need to avoid panicking here for when
+		// ex is of another base type
+		return nil
+	}
+	return exchange.RegisterMetrics()
+}

nodebuilder/header/service.go

@@ -9,7 +9,7 @@ import (
 	"github.com/celestiaorg/celestia-node/libs/header/sync"
 )
 
-// Service represents the header Service that can be started / stopped on a node.
+// Service represents the header service that can be started / stopped on a node.
 // Service's main function is to manage its sub-services. Service can contain several
 // sub-services, such as Exchange, ExchangeServer, Syncer, and so forth.
 type Service struct {
@@ -45,6 +45,11 @@ func (s *Service) Head(ctx context.Context) (*header.ExtendedHeader, error) {
 	return s.store.Head(ctx)
 }
 
-func (s *Service) IsSyncing(context.Context) bool {
+func (s *Service) IsSyncing(ctx context.Context) bool {
 	return !s.syncer.State().Finished()
 }
+
+// SyncerHead returns the ExtendedHeader of the chain head from the validator network.
+func (s *Service) SyncerHead(ctx context.Context) (*header.ExtendedHeader, error) {
+	return s.syncer.Head(ctx)
+}

nodebuilder/p2p/metrics.go

@@ -0,0 +1,82 @@
+package p2p
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/metrics"
+	"go.opentelemetry.io/otel/metric/global"
+	"go.opentelemetry.io/otel/metric/instrument"
+	"go.opentelemetry.io/otel/metric/unit"
+)
+
+// global meter provider (see opentelemetry docs)
+var (
+	meter = global.MeterProvider().Meter("p2p")
+)
+
+// WithMetrics option sets up metrics for p2p networking.
+func WithMetrics(bc *metrics.BandwidthCounter) error {
+	bandwidthTotalInbound, err := meter.
+		SyncInt64().
+		Histogram(
+			"p2p_bandwidth_total_inbound",
+			instrument.WithUnit(unit.Bytes),
+			instrument.WithDescription("total number of bytes received by the host"),
+		)
+	if err != nil {
+		return err
+	}
+	bandwidthTotalOutbound, _ := meter.
+		SyncInt64().
+		Histogram(
+			"p2p_bandwidth_total_outbound",
+			instrument.WithUnit(unit.Bytes),
+			instrument.WithDescription("total number of bytes sent by the host"),
+		)
+	if err != nil {
+		return err
+	}
+	bandwidthRateInbound, _ := meter.
+		SyncFloat64().
+		Histogram(
+			"p2p_bandwidth_rate_inbound",
+			instrument.WithDescription("total number of bytes sent by the host"),
+		)
+	if err != nil {
+		return err
+	}
+	bandwidthRateOutbound, _ := meter.
+		SyncFloat64().
+		Histogram(
+			"p2p_bandwidth_rate_outbound",
+			instrument.WithDescription("total number of bytes sent by the host"),
+		)
+	if err != nil {
+		return err
+	}
+	p2pPeerCount, _ := meter.
+		AsyncFloat64().
+		Gauge(
+			"p2p_peer_count",
+			instrument.WithDescription("number of peers connected to the host"),
+		)
+	if err != nil {
+		return err
+	}
+
+	return meter.RegisterCallback(
+		[]instrument.Asynchronous{
+			p2pPeerCount,
+		}, func(ctx context.Context) {
+			bcStats := bc.GetBandwidthTotals()
+			bcByPeerStats := bc.GetBandwidthByPeer()
+
+			bandwidthTotalInbound.Record(ctx, bcStats.TotalIn)
+			bandwidthTotalOutbound.Record(ctx, bcStats.TotalOut)
+			bandwidthRateInbound.Record(ctx, bcStats.RateIn)
+			bandwidthRateOutbound.Record(ctx, bcStats.RateOut)
+
+			p2pPeerCount.Observe(ctx, float64(len(bcByPeerStats)))
+		},
+	)
+}

nodebuilder/settings.go

@@ -15,10 +15,12 @@ import (
 
 	fraudPkg "github.com/celestiaorg/celestia-node/fraud"
 	headerPkg "github.com/celestiaorg/celestia-node/header"
+	header "github.com/celestiaorg/celestia-node/nodebuilder/header"
 
 	"github.com/celestiaorg/celestia-node/nodebuilder/das"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 	"github.com/celestiaorg/celestia-node/nodebuilder/p2p"
+	sharePkg "github.com/celestiaorg/celestia-node/nodebuilder/share"
 	"github.com/celestiaorg/celestia-node/state"
 )
 
@@ -39,10 +41,12 @@ func WithMetrics(metricOpts []otlpmetrichttp.Option, nodeType node.Type) fx.Opti
 	baseComponents := fx.Options(
 		fx.Supply(metricOpts),
 		fx.Invoke(initializeMetrics),
-		fx.Invoke(headerPkg.WithMetrics),
 		fx.Invoke(state.WithMetrics),
+		fx.Invoke(p2p.WithMetrics),
 		fx.Invoke(fraudPkg.WithMetrics),
 		fx.Invoke(node.WithMetrics),
+		fx.Invoke(headerPkg.WithMetrics),
+		fx.Invoke(header.WithMetrics),
 	)
 
 	var opts fx.Option
@@ -64,6 +68,16 @@ func WithMetrics(metricOpts []otlpmetrichttp.Option, nodeType node.Type) fx.Opti
 	return opts
 }
 
+// WithBlackboxMetrics enables blackbox metrics for the node.
+// Blackbox metrics are metrics that are recorded for the node's components
+// through a proxy that records metrics for the node's components
+// on each method call.
+func WithBlackboxMetrics() fx.Option {
+	return fx.Options(
+		fx.Decorate(sharePkg.WithBlackBoxMetrics),
+	)
+}
+
 // initializeMetrics initializes the global meter provider.
 func initializeMetrics(
 	ctx context.Context,