feat(metrics): implement AddNetworkTag function for consistent networ…

…k tagging across multiple modules
filecoin-project · Dec 12, 2024 · 775cfac · 775cfac
1 parent 65a02f6
commit 775cfac
Show file tree

Hide file tree

Showing 9 changed files with 29 additions and 42 deletions.
diff --git a/metrics/metrics.go b/metrics/metrics.go
@@ -11,6 +11,7 @@ import (
 	rpcmetrics "github.com/filecoin-project/go-jsonrpc/metrics"
 
 	"github.com/filecoin-project/lotus/blockstore"
+	"github.com/filecoin-project/lotus/build/buildconstants"
 )
 
 // Distribution
@@ -851,3 +852,8 @@ func Timer(ctx context.Context, m *stats.Float64Measure) func() time.Duration {
 		return time.Since(start)
 	}
 }
+
+func AddNetworkTag(ctx context.Context) context.Context {
+	ctx, _ = tag.New(ctx, tag.Upsert(Network, buildconstants.NetworkBundle))
+	return ctx
+}
diff --git a/node/modules/lp2p/rcmgr.go b/node/modules/lp2p/rcmgr.go
@@ -196,7 +196,7 @@ func (r rcmgrMetrics) AllowConn(dir network.Direction, usefd bool) {
 
 func (r rcmgrMetrics) BlockConn(dir network.Direction, usefd bool) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	if dir == network.DirInbound {
 		ctx, _ = tag.New(ctx, tag.Upsert(metrics.Direction, "inbound"))
 	} else {
@@ -212,7 +212,7 @@ func (r rcmgrMetrics) BlockConn(dir network.Direction, usefd bool) {
 
 func (r rcmgrMetrics) AllowStream(p peer.ID, dir network.Direction) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	if dir == network.DirInbound {
 		ctx, _ = tag.New(ctx, tag.Upsert(metrics.Direction, "inbound"))
 	} else {
@@ -223,7 +223,7 @@ func (r rcmgrMetrics) AllowStream(p peer.ID, dir network.Direction) {
 
 func (r rcmgrMetrics) BlockStream(p peer.ID, dir network.Direction) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	if dir == network.DirInbound {
 		ctx, _ = tag.New(ctx, tag.Upsert(metrics.Direction, "inbound"))
 	} else {
@@ -234,54 +234,54 @@ func (r rcmgrMetrics) BlockStream(p peer.ID, dir network.Direction) {
 
 func (r rcmgrMetrics) AllowPeer(p peer.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	stats.Record(ctx, metrics.RcmgrAllowPeer.M(1))
 }
 
 func (r rcmgrMetrics) BlockPeer(p peer.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	stats.Record(ctx, metrics.RcmgrBlockPeer.M(1))
 }
 
 func (r rcmgrMetrics) AllowProtocol(proto protocol.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ProtocolID, string(proto)))
 	stats.Record(ctx, metrics.RcmgrAllowProto.M(1))
 }
 
 func (r rcmgrMetrics) BlockProtocol(proto protocol.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ProtocolID, string(proto)))
 	stats.Record(ctx, metrics.RcmgrBlockProto.M(1))
 }
 
 func (r rcmgrMetrics) BlockProtocolPeer(proto protocol.ID, p peer.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ProtocolID, string(proto)))
 	stats.Record(ctx, metrics.RcmgrBlockProtoPeer.M(1))
 }
 
 func (r rcmgrMetrics) AllowService(svc string) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ServiceID, svc))
 	stats.Record(ctx, metrics.RcmgrAllowSvc.M(1))
 }
 
 func (r rcmgrMetrics) BlockService(svc string) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ServiceID, svc))
 	stats.Record(ctx, metrics.RcmgrBlockSvc.M(1))
 }
 
 func (r rcmgrMetrics) BlockServicePeer(svc string, p peer.ID) {
 	ctx := context.Background()
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+	ctx = metrics.AddNetworkTag(ctx)
 	ctx, _ = tag.New(ctx, tag.Upsert(metrics.ServiceID, svc))
 	stats.Record(ctx, metrics.RcmgrBlockSvcPeer.M(1))
 }

diff --git a/node/modules/paych.go b/node/modules/paych.go
@@ -8,6 +8,7 @@ import (
 	"go.uber.org/fx"
 
 	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node/impl/full"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -17,7 +18,7 @@ import (
 func NewManager(mctx helpers.MetricsCtx, lc fx.Lifecycle, sm stmgr.StateManagerAPI, pchstore *paychmgr.Store, api paychmgr.PaychAPI) *paychmgr.Manager {
 	ctx := helpers.LifecycleCtx(mctx, lc)
 	ctx, shutdown := context.WithCancel(ctx)
-
+	ctx = metrics.AddNetworkTag(ctx)
 	return paychmgr.NewManager(ctx, shutdown, sm, pchstore, api)
 }
 

diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go
@@ -11,6 +11,7 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	"github.com/jpillora/backoff"
+	"go.opencensus.io/tag"
 	"go.uber.org/fx"
 	"go.uber.org/multierr"
 	"golang.org/x/xerrors"
@@ -33,6 +34,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/lf3"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/journal"
+	"github.com/filecoin-project/lotus/metrics"
 	lotusminer "github.com/filecoin-project/lotus/miner"
 	"github.com/filecoin-project/lotus/node/config"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -289,7 +291,7 @@ func WindowPostScheduler(fc config.MinerFeeConfig, pc config.ProvingConfig) func
 		)
 
 		ctx := helpers.LifecycleCtx(mctx, lc)
-
+		ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
 		fps, err := wdpost.NewWindowedPoStScheduler(api, fc, pc, as, sealer, verif, sealer, j, []dtypes.MinerAddress{params.Maddr})
 
 		if err != nil {

diff --git a/node/rpc.go b/node/rpc.go
@@ -22,7 +22,6 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/v0api"
 	"github.com/filecoin-project/lotus/api/v1api"
-	"github.com/filecoin-project/lotus/build/buildconstants"
 	"github.com/filecoin-project/lotus/lib/rpcenc"
 	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/metrics/proxy"
@@ -50,7 +49,7 @@ func ServeRPC(h http.Handler, id string, addr multiaddr.Multiaddr) (StopFunc, er
 		ReadHeaderTimeout: 30 * time.Second,
 		BaseContext: func(listener net.Listener) context.Context {
 			ctx := context.Background()
-			ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
+			ctx = metrics.AddNetworkTag(ctx)
 			ctx, _ = tag.New(ctx, tag.Upsert(metrics.APIInterface, id))
 			return ctx
 		},

diff --git a/paychmgr/manager.go b/paychmgr/manager.go
@@ -73,10 +73,6 @@ type Manager struct {
 
 func NewManager(ctx context.Context, shutdown func(), sm stmgr.StateManagerAPI, pchstore *Store, api PaychAPI) *Manager {
 	impl := &managerAPIImpl{StateManagerAPI: sm, PaychAPI: api}
-
-	// Add network tag to context
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
-
 	return &Manager{
 		ctx:      ctx,
 		shutdown: shutdown,

diff --git a/storage/paths/db_index.go b/storage/paths/db_index.go
@@ -17,7 +17,6 @@ import (
 
 	"github.com/filecoin-project/go-state-types/abi"
 
-	"github.com/filecoin-project/lotus/build/buildconstants"
 	"github.com/filecoin-project/lotus/journal/alerting"
 	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
 	"github.com/filecoin-project/lotus/metrics"
@@ -47,12 +46,6 @@ func NewDBIndex(al *alerting.Alerting, db *harmonydb.DB) *DBIndex {
 	}
 }
 
-// addNetworkTag adds the network tag to the context for metrics
-func addNetworkTag(ctx context.Context) context.Context {
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
-	return ctx
-}
-
 func (dbi *DBIndex) StorageList(ctx context.Context) (map[storiface.ID][]storiface.Decl, error) {
 
 	var sectorEntries []struct {
@@ -127,7 +120,7 @@ func splitString(str string) []string {
 }
 
 func (dbi *DBIndex) StorageAttach(ctx context.Context, si storiface.StorageInfo, st fsutil.FsStat) error {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 
 	var allow, deny = make([]string, 0, len(si.AllowTypes)), make([]string, 0, len(si.DenyTypes))
 
@@ -321,7 +314,7 @@ func (dbi *DBIndex) StorageDetach(ctx context.Context, id storiface.ID, url stri
 }
 
 func (dbi *DBIndex) StorageReportHealth(ctx context.Context, id storiface.ID, report storiface.HealthReport) error {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 
 	retryWait := time.Millisecond * 20
 retryReportHealth:
@@ -389,7 +382,7 @@ func (dbi *DBIndex) checkFileType(fileType storiface.SectorFileType) bool {
 }
 
 func (dbi *DBIndex) StorageDeclareSector(ctx context.Context, storageID storiface.ID, s abi.SectorID, ft storiface.SectorFileType, primary bool) error {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 
 	if !dbi.checkFileType(ft) {
 		return xerrors.Errorf("invalid filetype")
@@ -690,7 +683,7 @@ func (dbi *DBIndex) StorageInfo(ctx context.Context, id storiface.ID) (storiface
 }
 
 func (dbi *DBIndex) StorageBestAlloc(ctx context.Context, allocate storiface.SectorFileType, ssize abi.SectorSize, pathType storiface.PathType, miner abi.ActorID) ([]storiface.StorageInfo, error) {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 
 	var err error
 	var spaceReq uint64

diff --git a/storage/wdpost/wdpost_changehandler.go b/storage/wdpost/wdpost_changehandler.go
@@ -4,13 +4,10 @@ import (
 	"context"
 	"sync"
 
-	"go.opencensus.io/tag"
-
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/dline"
 
-	"github.com/filecoin-project/lotus/build/buildconstants"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/metrics"
@@ -55,7 +52,6 @@ func (ch *changeHandler) start() {
 }
 
 func (ch *changeHandler) update(ctx context.Context, revert *types.TipSet, advance *types.TipSet) error {
-	ctx = addNetworkTag(ctx)
 	// Get the current deadline period
 	di, err := ch.api.StateMinerProvingDeadline(ctx, ch.actor, advance.Key())
 	if err != nil {
@@ -215,7 +211,7 @@ func (p *proveHandler) run() {
 }
 
 func (p *proveHandler) processHeadChange(ctx context.Context, newTS *types.TipSet, di *dline.Info) {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 	// If the post window has expired, abort the current proof
 	if p.current != nil && newTS.Height() >= p.current.di.Close {
 		// Cancel the context on the current proof
@@ -393,7 +389,7 @@ func (s *submitHandler) run() {
 
 // processHeadChange is called when the chain head changes
 func (s *submitHandler) processHeadChange(ctx context.Context, revert *types.TipSet, advance *types.TipSet, di *dline.Info) {
-	ctx = addNetworkTag(ctx)
+	ctx = metrics.AddNetworkTag(ctx)
 	s.currentCtx = ctx
 	s.currentTS = advance
 	s.currentDI = di
@@ -547,8 +543,3 @@ func NextDeadline(currentDeadline *dline.Info) *dline.Info {
 func NewDeadlineInfo(periodStart abi.ChainEpoch, deadlineIdx uint64, currEpoch abi.ChainEpoch) *dline.Info {
 	return dline.NewInfo(periodStart, deadlineIdx, currEpoch, miner.WPoStPeriodDeadlines, miner.WPoStProvingPeriod(), miner.WPoStChallengeWindow(), miner.WPoStChallengeLookback, miner.FaultDeclarationCutoff)
 }
-
-func addNetworkTag(ctx context.Context) context.Context {
-	ctx, _ = tag.New(ctx, tag.Upsert(metrics.Network, buildconstants.NetworkBundle))
-	return ctx
-}
diff --git a/storage/wdpost/wdpost_sched.go b/storage/wdpost/wdpost_sched.go
@@ -190,7 +190,6 @@ func (s *WindowPoStScheduler) Run(ctx context.Context) {
 				}
 
 				ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.headChange")
-
 				s.update(ctx, nil, chg.Val)
 
 				span.End()
-Original file line number
+Diff line change
@@ Expand Up / @@ -190,7 +190,6 @@ func (s *WindowPoStScheduler) Run(ctx context.Context) { @@
     				}
     				ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.headChange")
     				s.update(ctx, nil, chg.Val)
     				span.End()
@@ Expand Down @@