Merge pull request #499 from ipfs-force-community/feat/add-more-metrics

Feat/add more metrics

.golangci.yml

@@ -43,6 +43,10 @@ issues:
 linters-settings:
   goconst:
     min-occurrences: 6
+  revive:
+    rules:
+      - name: unused-parameter
+        disabled: true
 
 run:
   skip-dirs-use-default: false

dagstore/wrapper.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/filecoin-project/go-statemachine/fsm"
 	"github.com/ipfs-force-community/droplet/v2/config"
+	"github.com/ipfs-force-community/droplet/v2/metrics"
 	"github.com/ipfs-force-community/droplet/v2/models/badger"
 	"github.com/ipfs-force-community/droplet/v2/models/repo"
 	carindex "github.com/ipld/go-car/v2/index"
@@ -127,6 +128,30 @@ func NewDAGStore(ctx context.Context,
 		return nil, nil, fmt.Errorf("failed to create DAG store: %w", err)
 	}
 
+	// thread for metrics
+	go func() {
+		tick := time.NewTicker(1 * time.Minute)
+		defer tick.Stop()
+		for {
+			select {
+			case <-tick.C:
+				infos := dagst.AllShardsInfo()
+				stateCount := make(map[dagstore.ShardState]int)
+				for _, info := range infos {
+					if _, ok := stateCount[info.ShardState]; !ok {
+						stateCount[info.ShardState] = 0
+					}
+					stateCount[info.ShardState]++
+				}
+				for state, count := range stateCount {
+					metrics.ShardNum.Set(ctx, state.String(), int64(count))
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
 	w := &Wrapper{
 		cfg:        cfg,
 		dagst:      dagst,

go.mod

@@ -37,7 +37,7 @@ require (
 	github.com/hannahhoward/go-pubsub v0.0.0-20200423002714-8d62886cc36e
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/howeyc/gopass v0.0.0-20210920133722-c8aef6fb66ef
-	github.com/ipfs-force-community/metrics v1.0.1-0.20231011024528-8c881d456601
+	github.com/ipfs-force-community/metrics v1.0.1-0.20231207081445-30178e706d09
 	github.com/ipfs-force-community/sophon-auth v1.14.0
 	github.com/ipfs-force-community/sophon-gateway v1.14.0
 	github.com/ipfs-force-community/sophon-messager v1.14.0

go.sum

@@ -698,8 +698,8 @@ github.com/ipfs-force-community/go-fil-markets v1.2.6-0.20230822060005-aee2cbae5
 github.com/ipfs-force-community/go-fil-markets v1.2.6-0.20230822060005-aee2cbae5b01/go.mod h1:eryxo/oVgIxaR5g5CNr9PlvZOi+u/bak0IsPL/PT1hk=
 github.com/ipfs-force-community/go-jsonrpc v0.1.9 h1:5QavBltfvV6fz/+EbYsCkVxJ1MSJncZm6YuPs1SLdZU=
 github.com/ipfs-force-community/go-jsonrpc v0.1.9/go.mod h1:jBSvPTl8V1N7gSTuCR4bis8wnQnIjHbRPpROol6iQKM=
-github.com/ipfs-force-community/metrics v1.0.1-0.20231011024528-8c881d456601 h1:zxKQ30KAD6KfvSFAx9tuqQXLDsEHyF+eVaUBXXYC2bU=
-github.com/ipfs-force-community/metrics v1.0.1-0.20231011024528-8c881d456601/go.mod h1:wM6EmkEcnJgWOFcVytgvK0u15awEmt8He0f2kAdsFDA=
+github.com/ipfs-force-community/metrics v1.0.1-0.20231207081445-30178e706d09 h1:qEI6ItxKtgOupMMuGJwqK5zEzztKKPUP1QKq9g+X5bM=
+github.com/ipfs-force-community/metrics v1.0.1-0.20231207081445-30178e706d09/go.mod h1:wM6EmkEcnJgWOFcVytgvK0u15awEmt8He0f2kAdsFDA=
 github.com/ipfs-force-community/sophon-auth v1.14.0 h1:ctBJ6UHkcytEzfVPgiiHo0cW4FGQrE7r1H3Um0FcHbo=
 github.com/ipfs-force-community/sophon-auth v1.14.0/go.mod h1:d6J6u3zyIwcEajRho5BhVBcoIChEf0K76wP4yJEfEhc=
 github.com/ipfs-force-community/sophon-gateway v1.14.0 h1:nl4RnYiwVwviYbi0qhdisUHqHaOdoux1BbT6ps/F1iY=

metrics/metrics.go

@@ -1,7 +1,8 @@
 package metrics
 
 import (
-	"github.com/filecoin-project/go-jsonrpc/metrics"
+	rpcMetrics "github.com/filecoin-project/go-jsonrpc/metrics"
+	"github.com/ipfs-force-community/metrics"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
@@ -12,6 +13,12 @@ var (
 	StorageNameTag, _ = tag.NewKey("storage")
 )
 
+var (
+	ApiState               = metrics.NewInt64("api/state", "api service state. 0: down, 1: up", "")
+	RetrievalTransferEvent = metrics.NewCounterWithCategory("retrieval/transfer_event", "retrieval transfer event")
+	ShardNum               = metrics.NewInt64WithCategory("shard/num", "shard num in different state", "")
+)
+
 var (
 	GraphsyncReceivingPeersCount              = stats.Int64("graphsync/receiving_peers", "number of peers we are receiving graphsync data from", stats.UnitDimensionless)
 	GraphsyncReceivingActiveCount             = stats.Int64("graphsync/receiving_active", "number of active receiving graphsync transfers", stats.UnitDimensionless)
@@ -127,4 +134,12 @@ var views = append([]*view.View{
 
 	StorageRetrievalHitCountView,
 	StorageSaveHitCountView,
-}, metrics.DefaultViews...)
+}, rpcMetrics.DefaultViews...)
+
+func init() {
+	for _, v := range views {
+		if err := view.Register(v); err != nil {
+			panic(err)
+		}
+	}
+}

metrics/modules.go

@@ -18,7 +18,8 @@ var MetricsOpts = func(scope string, metricsConfig *metrics.MetricsConfig) build
 			return metrics2.CtxScope(context.Background(), scope)
 		}),
 		builder.Override(startMetricsKey, func(mctx metrics.MetricsCtx, lc fx.Lifecycle) error {
-			return SetupMetrics(metrics.LifecycleCtx(mctx, lc), metricsConfig)
+			ctx := metrics.LifecycleCtx(mctx, lc)
+			return metrics.SetupMetrics(ctx, metricsConfig)
 		}),
 	)
 }

retrievalprovider/provider_datatransfer_sub.go

@@ -6,6 +6,7 @@ import (
 
 	datatransfer "github.com/filecoin-project/go-data-transfer/v2"
 	rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	"github.com/ipfs-force-community/droplet/v2/metrics"
 )
 
 // ProviderDataTransferSubscriber is the function called when an event occurs in a data
@@ -39,6 +40,8 @@ func ProviderDataTransferSubscriber(deals IDatatransferHandler) datatransfer.Sub
 			}
 		}
 
+		metrics.RetrievalTransferEvent.Tick(ctx, event.Code.String())
+
 		switch event.Code {
 		case datatransfer.Accept:
 			mlog = mlog.With("retrievalEvent", rm.ProviderEvents[rm.ProviderEventDealAccepted])

rpc/rpc.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ipfs-force-community/sophon-auth/jwtclient"
 
 	"github.com/ipfs-force-community/droplet/v2/config"
+	"github.com/ipfs-force-community/droplet/v2/metrics"
 	"github.com/ipfs-force-community/droplet/v2/retrievalprovider/httpretrieval"
 )
 
@@ -89,6 +90,7 @@ func ServeRPC(
 			log.Errorf("shutting down RPC server failed: %s", err)
 		}
 		log.Warn("RPC Graceful shutdown successful")
+		metrics.ApiState.Set(ctx, 0)
 	}()
 
 	addr, err := multiaddr.NewMultiaddr(apiCfg.ListenAddress)
@@ -100,6 +102,7 @@ func ServeRPC(
 	if err != nil {
 		return err
 	}
+	metrics.ApiState.Set(ctx, 1)
 	log.Infof("start rpc listen %s", addr)
 
 	if err := srv.Serve(manet.NetListener(nl)); err != nil && err != http.ErrServerClosed {