Merge pull request #7329 from filecoin-project/feat/lotus-stats-update

Add caches to lotus-stats and splitcode

Dockerfile.lotus

@@ -36,7 +36,7 @@ WORKDIR /opt/filecoin
 ARG RUSTFLAGS=""
 ARG GOFLAGS=""
 
-RUN make lotus lotus-miner lotus-worker lotus-shed lotus-wallet lotus-gateway
+RUN make lotus lotus-miner lotus-worker lotus-shed lotus-wallet lotus-gateway lotus-stats
 
 
 FROM ubuntu:20.04 AS base
@@ -183,6 +183,7 @@ COPY --from=builder /opt/filecoin/lotus-wallet /usr/local/bin/
 COPY --from=builder /opt/filecoin/lotus-gateway /usr/local/bin/
 COPY --from=builder /opt/filecoin/lotus-miner /usr/local/bin/
 COPY --from=builder /opt/filecoin/lotus-worker /usr/local/bin/
+COPY --from=builder /opt/filecoin/lotus-stats /usr/local/bin/
 
 RUN mkdir /var/tmp/filecoin-proof-parameters
 RUN mkdir /var/lib/lotus

cmd/lotus-pcr/main.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin"
+	lcli "github.com/filecoin-project/lotus/cli"
 
 	miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
 
@@ -41,7 +42,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/actors/builtin/market"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
 	"github.com/filecoin-project/lotus/chain/types"
-	"github.com/filecoin-project/lotus/tools/stats"
+	"github.com/filecoin-project/lotus/tools/stats/sync"
 )
 
 var log = logging.Logger("main")
@@ -160,15 +161,15 @@ var findMinersCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()
-		api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
+		api, closer, err := lcli.GetFullNodeAPI(cctx)
 		if err != nil {
-			log.Fatal(err)
+			return err
 		}
 		defer closer()
 
 		if !cctx.Bool("no-sync") {
-			if err := stats.WaitForSyncComplete(ctx, api); err != nil {
-				log.Fatal(err)
+			if err := sync.SyncWait(ctx, api); err != nil {
+				return err
 			}
 		}
 
@@ -245,7 +246,7 @@ var recoverMinersCmd = &cli.Command{
 	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()
-		api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
+		api, closer, err := lcli.GetFullNodeAPI(cctx)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -266,8 +267,8 @@ var recoverMinersCmd = &cli.Command{
 		}
 
 		if !cctx.Bool("no-sync") {
-			if err := stats.WaitForSyncComplete(ctx, api); err != nil {
-				log.Fatal(err)
+			if err := sync.SyncWait(ctx, api); err != nil {
+				return err
 			}
 		}
 
@@ -427,7 +428,7 @@ var runCmd = &cli.Command{
 		}()
 
 		ctx := context.Background()
-		api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
+		api, closer, err := lcli.GetFullNodeAPI(cctx)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -448,12 +449,12 @@ var runCmd = &cli.Command{
 		}
 
 		if !cctx.Bool("no-sync") {
-			if err := stats.WaitForSyncComplete(ctx, api); err != nil {
-				log.Fatal(err)
+			if err := sync.SyncWait(ctx, api); err != nil {
+				return err
 			}
 		}
 
-		tipsetsCh, err := stats.GetTips(ctx, api, r.Height(), cctx.Int("head-delay"))
+		tipsetsCh, err := sync.BufferedTipsetChannel(ctx, api, r.Height(), cctx.Int("head-delay"))
 		if err != nil {
 			log.Fatal(err)
 		}

cmd/lotus-stats/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3'
 
 services:
   influxdb:
-    image: influxdb:latest
+    image: influxdb:1.8
     container_name: influxdb
     ports:
       - "18086:8086"

cmd/lotus-stats/main.go

@@ -2,18 +2,36 @@ package main
 
 import (
 	"context"
+	"net/http"
+	_ "net/http/pprof"
 	"os"
+	"time"
 
+	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/build"
 	lcli "github.com/filecoin-project/lotus/cli"
-	"github.com/filecoin-project/lotus/tools/stats"
+	"github.com/filecoin-project/lotus/tools/stats/influx"
+	"github.com/filecoin-project/lotus/tools/stats/ipldstore"
+	"github.com/filecoin-project/lotus/tools/stats/metrics"
+	"github.com/filecoin-project/lotus/tools/stats/points"
+	"github.com/filecoin-project/lotus/tools/stats/sync"
 
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/urfave/cli/v2"
+
+	"contrib.go.opencensus.io/exporter/prometheus"
+	stats "go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
 )
 
 var log = logging.Logger("stats")
 
+func init() {
+	if err := view.Register(metrics.DefaultViews...); err != nil {
+		log.Fatal(err)
+	}
+}
+
 func main() {
 	local := []*cli.Command{
 		runCmd,
@@ -37,7 +55,7 @@ func main() {
 			},
 		},
 		Before: func(cctx *cli.Context) error {
-			return logging.SetLogLevel("stats", cctx.String("log-level"))
+			return logging.SetLogLevelRegex("stats/*", cctx.String("log-level"))
 		},
 		Commands: local,
 	}
@@ -104,6 +122,12 @@ var runCmd = &cli.Command{
 			Usage:   "do not wait for chain sync to complete",
 			Value:   false,
 		},
+		&cli.IntFlag{
+			Name:    "ipld-store-cache-size",
+			Usage:   "size of lru cache for ChainReadObj",
+			EnvVars: []string{"LOTUS_STATS_IPLD_STORE_CACHE_SIZE"},
+			Value:   2 << 15,
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()
@@ -118,43 +142,125 @@ var runCmd = &cli.Command{
 		influxPasswordFlag := cctx.String("influx-password")
 		influxDatabaseFlag := cctx.String("influx-database")
 
+		ipldStoreCacheSizeFlag := cctx.Int("ipld-store-cache-size")
+
 		log.Infow("opening influx client", "hostname", influxHostnameFlag, "username", influxUsernameFlag, "database", influxDatabaseFlag)
 
-		influx, err := stats.InfluxClient(influxHostnameFlag, influxUsernameFlag, influxPasswordFlag)
+		influxClient, err := influx.NewClient(influxHostnameFlag, influxUsernameFlag, influxPasswordFlag)
 		if err != nil {
-			log.Fatal(err)
+			return err
 		}
 
+		exporter, err := prometheus.NewExporter(prometheus.Options{
+			Namespace: "lotus_stats",
+		})
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			http.Handle("/metrics", exporter)
+			if err := http.ListenAndServe(":6688", nil); err != nil {
+				log.Errorw("failed to start http server", "err", err)
+			}
+		}()
+
 		if resetFlag {
-			if err := stats.ResetDatabase(influx, influxDatabaseFlag); err != nil {
-				log.Fatal(err)
+			if err := influx.ResetDatabase(influxClient, influxDatabaseFlag); err != nil {
+				return err
 			}
 		}
 
-		height := int64(heightFlag)
+		api, closer, err := lcli.GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		if !noSyncFlag {
+			if err := sync.SyncWait(ctx, api); err != nil {
+				return err
+			}
+		}
+
+		gtp, err := api.ChainGetGenesis(ctx)
+		if err != nil {
+			return err
+		}
 
+		genesisTime := time.Unix(int64(gtp.MinTimestamp()), 0)
+
+		// When height is set to `0` we will resume from the best height we can.
+		// The goal is to ensure we have data in the last 60 tipsets
+		height := int64(heightFlag)
 		if !resetFlag && height == 0 {
-			h, err := stats.GetLastRecordedHeight(influx, influxDatabaseFlag)
+			lastHeight, err := influx.GetLastRecordedHeight(influxClient, influxDatabaseFlag)
+			if err != nil {
+				return err
+			}
+
+			sinceGenesis := build.Clock.Now().Sub(genesisTime)
+			expectedHeight := int64(sinceGenesis.Seconds()) / int64(build.BlockDelaySecs)
+
+			startOfWindowHeight := expectedHeight - 60
+
+			if lastHeight > startOfWindowHeight {
+				height = lastHeight
+			} else {
+				height = startOfWindowHeight
+			}
+
+			ts, err := api.ChainHead(ctx)
 			if err != nil {
-				log.Info(err)
+				return err
 			}
 
-			height = h
+			headHeight := int64(ts.Height())
+			if headHeight < height {
+				height = headHeight
+			}
 		}
 
-		api, closer, err := lcli.GetFullNodeAPI(cctx)
+		go func() {
+			t := time.NewTicker(time.Second)
+
+			for {
+				select {
+				case <-t.C:
+					sinceGenesis := build.Clock.Now().Sub(genesisTime)
+					expectedHeight := int64(sinceGenesis.Seconds()) / int64(build.BlockDelaySecs)
+
+					stats.Record(ctx, metrics.TipsetCollectionHeightExpected.M(expectedHeight))
+				}
+			}
+		}()
+
+		store, err := ipldstore.NewApiIpldStore(ctx, api, ipldStoreCacheSizeFlag)
 		if err != nil {
 			return err
 		}
-		defer closer()
 
-		if !noSyncFlag {
-			if err := stats.WaitForSyncComplete(ctx, api); err != nil {
-				log.Fatal(err)
-			}
+		collector, err := points.NewChainPointCollector(ctx, store, api)
+		if err != nil {
+			return err
 		}
 
-		stats.Collect(ctx, api, influx, influxDatabaseFlag, height, headLagFlag)
+		tipsets, err := sync.BufferedTipsetChannel(ctx, api, abi.ChainEpoch(height), headLagFlag)
+		if err != nil {
+			return err
+		}
+
+		wq := influx.NewWriteQueue(ctx, influxClient)
+		defer wq.Close()
+
+		for tipset := range tipsets {
+			if nb, err := collector.Collect(ctx, tipset); err != nil {
+				log.Warnw("failed to collect points", "err", err)
+			} else {
+				nb.SetDatabase(influxDatabaseFlag)
+				wq.AddBatch(nb)
+			}
+		}
 
 		return nil
 	},

testplans/lotus-soup/go.mod

@@ -3,14 +3,14 @@ module github.com/filecoin-project/lotus/testplans/lotus-soup
 go 1.16
 
 require (
-	contrib.go.opencensus.io/exporter/prometheus v0.1.0
+	contrib.go.opencensus.io/exporter/prometheus v0.4.0
 	github.com/codeskyblue/go-sh v0.0.0-20200712050446-30169cf553fe
 	github.com/davecgh/go-spew v1.1.1
 	github.com/drand/drand v1.2.1
-	github.com/filecoin-project/go-address v0.0.5
-	github.com/filecoin-project/go-data-transfer v1.10.1
-	github.com/filecoin-project/go-fil-markets v1.12.0
-	github.com/filecoin-project/go-jsonrpc v0.1.4-0.20210217175800-45ea43ac2bec
+	github.com/filecoin-project/go-address v0.0.6
+	github.com/filecoin-project/go-data-transfer v1.11.4
+	github.com/filecoin-project/go-fil-markets v1.13.3
+	github.com/filecoin-project/go-jsonrpc v0.1.5
 	github.com/filecoin-project/go-state-types v0.1.1-0.20210915140513-d354ccf10379
 	github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b
 	github.com/filecoin-project/lotus v0.0.0-00010101000000-000000000000
@@ -21,17 +21,17 @@ require (
 	github.com/influxdata/influxdb v1.9.4 // indirect
 	github.com/ipfs/go-cid v0.1.0
 	github.com/ipfs/go-datastore v0.4.6
-	github.com/ipfs/go-ipfs-files v0.0.8
+	github.com/ipfs/go-ipfs-files v0.0.9
 	github.com/ipfs/go-ipld-format v0.2.0
 	github.com/ipfs/go-log/v2 v2.3.0
-	github.com/ipfs/go-merkledag v0.3.2
+	github.com/ipfs/go-merkledag v0.4.1
 	github.com/ipfs/go-unixfs v0.2.6
-	github.com/ipld/go-car v0.3.1-null-padded-files
+	github.com/ipld/go-car v0.3.2-0.20211001225732-32d0d9933823
 	github.com/kpacha/opencensus-influxdb v0.0.0-20181102202715-663e2683a27c
 	github.com/libp2p/go-libp2p v0.15.0
 	github.com/libp2p/go-libp2p-core v0.9.0
 	github.com/libp2p/go-libp2p-pubsub-tracer v0.0.0-20200626141350-e730b32bf1e6
-	github.com/multiformats/go-multiaddr v0.4.0
+	github.com/multiformats/go-multiaddr v0.4.1
 	github.com/testground/sdk-go v0.2.6
 	go.opencensus.io v0.23.0
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c