cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Add caches to lotus-stats and splitcode

Browse Source
This commit is contained in:
Travis Person
2021-11-01 09:05:14 +00:00
parent e0a9cae386
commit 2d4f5958e2
25 changed files with 1392 additions and 924 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/stats/collect.go
-63
View File
@@ -1,63 +0,0 @@
package stats
import (
"context"
"time"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api/v0api"
client "github.com/influxdata/influxdb1-client/v2"
)
func Collect(ctx context.Context, api v0api.FullNode, influx client.Client, database string, height int64, headlag int) {
tipsetsCh, err := GetTips(ctx, api, abi.ChainEpoch(height), headlag)
if err != nil {
log.Fatal(err)
}
wq := NewInfluxWriteQueue(ctx, influx)
defer wq.Close()
for tipset := range tipsetsCh {
log.Infow("Collect stats", "height", tipset.Height())
pl := NewPointList()
height := tipset.Height()
if err := RecordTipsetPoints(ctx, api, pl, tipset); err != nil {
log.Warnw("Failed to record tipset", "height", height, "error", err)
continue
}
if err := RecordTipsetMessagesPoints(ctx, api, pl, tipset); err != nil {
log.Warnw("Failed to record messages", "height", height, "error", err)
continue
}
if err := RecordTipsetStatePoints(ctx, api, pl, tipset); err != nil {
log.Warnw("Failed to record state", "height", height, "error", err)
continue
}
// Instead of having to pass around a bunch of generic stuff we want for each point
// we will just add them at the end.
tsTimestamp := time.Unix(int64(tipset.MinTimestamp()), int64(0))
nb, err := InfluxNewBatch()
if err != nil {
log.Fatal(err)
}
for _, pt := range pl.Points() {
pt.SetTime(tsTimestamp)
nb.AddPoint(NewPointFrom(pt))
}
nb.SetDatabase(database)
log.Infow("Adding points", "count", len(nb.Points()), "height", tipset.Height())
wq.AddBatch(nb)
}
}
tools/stats/head_buffer.go
-47
View File
@@ -1,47 +0,0 @@
package stats
import (
"container/list"
"github.com/filecoin-project/lotus/api"
)
type headBuffer struct {
buffer *list.List
size int
}
func newHeadBuffer(size int) *headBuffer {
buffer := list.New()
buffer.Init()
return &headBuffer{
buffer: buffer,
size: size,
}
}
func (h *headBuffer) push(hc *api.HeadChange) (rethc *api.HeadChange) {
if h.buffer.Len() == h.size {
var ok bool
el := h.buffer.Front()
rethc, ok = el.Value.(*api.HeadChange)
if !ok {
panic("Value from list is not the correct type")
}
h.buffer.Remove(el)
}
h.buffer.PushBack(hc)
return
}
func (h *headBuffer) pop() {
el := h.buffer.Back()
if el != nil {
h.buffer.Remove(el)
}
}
tools/stats/head_buffer_test.go
-43
View File
@@ -1,43 +0,0 @@
package stats
import (
"testing"
"github.com/filecoin-project/lotus/api"
"github.com/stretchr/testify/require"
)
func TestHeadBuffer(t *testing.T) {
t.Run("Straight push through", func(t *testing.T) {
hb := newHeadBuffer(5)
require.Nil(t, hb.push(&api.HeadChange{Type: "1"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "2"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "3"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "4"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "5"}))
hc := hb.push(&api.HeadChange{Type: "6"})
require.Equal(t, hc.Type, "1")
})
t.Run("Reverts", func(t *testing.T) {
hb := newHeadBuffer(5)
require.Nil(t, hb.push(&api.HeadChange{Type: "1"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "2"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "3"}))
hb.pop()
require.Nil(t, hb.push(&api.HeadChange{Type: "3a"}))
hb.pop()
require.Nil(t, hb.push(&api.HeadChange{Type: "3b"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "4"}))
require.Nil(t, hb.push(&api.HeadChange{Type: "5"}))
hc := hb.push(&api.HeadChange{Type: "6"})
require.Equal(t, hc.Type, "1")
hc = hb.push(&api.HeadChange{Type: "7"})
require.Equal(t, hc.Type, "2")
hc = hb.push(&api.HeadChange{Type: "8"})
require.Equal(t, hc.Type, "3b")
})
}
tools/stats/headbuffer/head_buffer.go
+56
View File
@@ -0,0 +1,56 @@
package headbuffer
import (
"container/list"
"github.com/filecoin-project/lotus/api"
)
type HeadChangeStackBuffer struct {
buffer *list.List
size int
}
// NewHeadChangeStackBuffer buffer HeadChange events to avoid having to
// deal with revert changes. Initialized size should be the average reorg
// size + 1
func NewHeadChangeStackBuffer(size int) *HeadChangeStackBuffer {
buffer := list.New()
buffer.Init()
return &HeadChangeStackBuffer{
buffer: buffer,
size: size,
}
}
// Push adds a HeadChange to stack buffer. If the length of
// the stack buffer grows larger than the initizlized size, the
// oldest HeadChange is returned.
func (h *HeadChangeStackBuffer) Push(hc *api.HeadChange) (rethc *api.HeadChange) {
if h.buffer.Len() >= h.size {
var ok bool
el := h.buffer.Front()
rethc, ok = el.Value.(*api.HeadChange)
if !ok {
// This shouldn't be possible, this method is typed and is the only place data
// pushed to the buffer.
panic("A cosmic ray made me do it")
}
h.buffer.Remove(el)
}
h.buffer.PushBack(hc)
return
}
// Pop removes the last added HeadChange
func (h *HeadChangeStackBuffer) Pop() {
el := h.buffer.Back()
if el != nil {
h.buffer.Remove(el)
}
}
tools/stats/headbuffer/head_buffer_test.go
+42
View File
@@ -0,0 +1,42 @@
package headbuffer
import (
"testing"
"github.com/filecoin-project/lotus/api"
"github.com/stretchr/testify/require"
)
func TestHeadBuffer(t *testing.T) {
t.Run("Straight Push through", func(t *testing.T) {
hb := NewHeadChangeStackBuffer(5)
require.Nil(t, hb.Push(&api.HeadChange{Type: "1"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "2"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "3"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "4"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "5"}))
hc := hb.Push(&api.HeadChange{Type: "6"})
require.Equal(t, hc.Type, "1")
})
t.Run("Reverts", func(t *testing.T) {
hb := NewHeadChangeStackBuffer(5)
require.Nil(t, hb.Push(&api.HeadChange{Type: "1"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "2"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "3"}))
hb.Pop()
require.Nil(t, hb.Push(&api.HeadChange{Type: "3a"}))
hb.Pop()
require.Nil(t, hb.Push(&api.HeadChange{Type: "3b"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "4"}))
require.Nil(t, hb.Push(&api.HeadChange{Type: "5"}))
hc := hb.Push(&api.HeadChange{Type: "6"})
require.Equal(t, hc.Type, "1")
hc = hb.Push(&api.HeadChange{Type: "7"})
require.Equal(t, hc.Type, "2")
hc = hb.Push(&api.HeadChange{Type: "8"})
require.Equal(t, hc.Type, "3b")
})
}
tools/stats/influx/influx.go
+133
View File
@@ -0,0 +1,133 @@
package influx
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/filecoin-project/lotus/build"
_ "github.com/influxdata/influxdb1-client"
models "github.com/influxdata/influxdb1-client/models"
client "github.com/influxdata/influxdb1-client/v2"
)
type PointList struct {
points []models.Point
}
func NewPointList() *PointList {
return &PointList{}
}
func (pl *PointList) AddPoint(p models.Point) {
pl.points = append(pl.points, p)
}
func (pl *PointList) Points() []models.Point {
return pl.points
}
type WriteQueue struct {
ch chan client.BatchPoints
}
func NewWriteQueue(ctx context.Context, influx client.Client) *WriteQueue {
ch := make(chan client.BatchPoints, 128)
maxRetries := 10
go func() {
main:
for {
select {
case <-ctx.Done():
return
case batch := <-ch:
for i := 0; i < maxRetries; i++ {
if err := influx.Write(batch); err != nil {
log.Warnw("Failed to write batch", "error", err)
build.Clock.Sleep(3 * time.Second)
continue
}
continue main
}
log.Error("dropping batch due to failure to write")
}
}
}()
return &WriteQueue{
ch: ch,
}
}
func (i *WriteQueue) AddBatch(bp client.BatchPoints) {
i.ch <- bp
}
func (i *WriteQueue) Close() {
close(i.ch)
}
func NewClient(addr, user, pass string) (client.Client, error) {
return client.NewHTTPClient(client.HTTPConfig{
Addr: addr,
Username: user,
Password: pass,
})
}
func NewBatch() (client.BatchPoints, error) {
return client.NewBatchPoints(client.BatchPointsConfig{})
}
func NewPoint(name string, value interface{}) models.Point {
pt, _ := models.NewPoint(name, models.Tags{},
map[string]interface{}{"value": value}, build.Clock.Now().UTC())
return pt
}
func NewPointFrom(p models.Point) *client.Point {
return client.NewPointFrom(p)
}
func ResetDatabase(influx client.Client, database string) error {
log.Debug("resetting database")
q := client.NewQuery(fmt.Sprintf(`DROP DATABASE "%s"; CREATE DATABASE "%s";`, database, database), "", "")
_, err := influx.Query(q)
if err != nil {
return err
}
log.Infow("database reset", "database", database)
return nil
}
func GetLastRecordedHeight(influx client.Client, database string) (int64, error) {
log.Debug("retrieving last record height")
q := client.NewQuery(`SELECT "value" FROM "chain.height" ORDER BY time DESC LIMIT 1`, database, "")
res, err := influx.Query(q)
if err != nil {
return 0, err
}
if len(res.Results) == 0 {
return 0, fmt.Errorf("No results found for last recorded height")
}
if len(res.Results[0].Series) == 0 {
return 0, fmt.Errorf("No results found for last recorded height")
}
height, err := (res.Results[0].Series[0].Values[0][1].(json.Number)).Int64()
if err != nil {
return 0, err
}
log.Infow("last record height", "height", height)
return height, nil
}
tools/stats/influx/log.go
+7
View File
@@ -0,0 +1,7 @@
package influx
import (
logging "github.com/ipfs/go-log/v2"
)
var log = logging.Logger("stats/influx")
tools/stats/ipldstore/ipldstore.go
+92
View File
@@ -0,0 +1,92 @@
package ipldstore
import (
"bytes"
"context"
"fmt"
"github.com/filecoin-project/lotus/tools/stats/metrics"
lru "github.com/hashicorp/golang-lru"
"github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
"go.opencensus.io/stats"
)
type ApiIpldStore struct {
ctx context.Context
api apiIpldStoreApi
cache *lru.TwoQueueCache
cacheSize int
}
type apiIpldStoreApi interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
}
func NewApiIpldStore(ctx context.Context, api apiIpldStoreApi, cacheSize int) (*ApiIpldStore, error) {
store := &ApiIpldStore{
ctx: ctx,
api: api,
cacheSize: cacheSize,
}
cache, err := lru.New2Q(store.cacheSize)
if err != nil {
return nil, err
}
store.cache = cache
return store, nil
}
func (ht *ApiIpldStore) Context() context.Context {
return ht.ctx
}
func (ht *ApiIpldStore) read(ctx context.Context, c cid.Cid) ([]byte, error) {
stats.Record(ctx, metrics.IpldStoreCacheMiss.M(1))
done := metrics.Timer(ctx, metrics.IpldStoreReadDuration)
defer done()
return ht.api.ChainReadObj(ctx, c)
}
func (ht *ApiIpldStore) Get(ctx context.Context, c cid.Cid, out interface{}) error {
done := metrics.Timer(ctx, metrics.IpldStoreGetDuration)
defer done()
defer func() {
stats.Record(ctx, metrics.IpldStoreCacheSize.M(int64(ht.cacheSize)))
stats.Record(ctx, metrics.IpldStoreCacheLength.M(int64(ht.cache.Len())))
}()
var raw []byte
if a, ok := ht.cache.Get(c); ok {
stats.Record(ctx, metrics.IpldStoreCacheHit.M(1))
raw = a.([]byte)
} else {
bs, err := ht.read(ctx, c)
if err != nil {
return err
}
raw = bs
}
cu, ok := out.(cbg.CBORUnmarshaler)
if ok {
if err := cu.UnmarshalCBOR(bytes.NewReader(raw)); err != nil {
return err
}
ht.cache.Add(c, raw)
return nil
}
return fmt.Errorf("Object does not implement CBORUnmarshaler")
}
func (ht *ApiIpldStore) Put(ctx context.Context, v interface{}) (cid.Cid, error) {
return cid.Undef, fmt.Errorf("Put is not implemented on ApiIpldStore")
}
tools/stats/metrics.go
-418
View File
@@ -1,418 +0,0 @@
package stats
import (
"bytes"
"context"
"encoding/json"
"fmt"
"math"
"math/big"
"strings"
"time"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors/builtin/power"
"github.com/filecoin-project/lotus/chain/actors/builtin/reward"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/ipfs/go-cid"
"github.com/multiformats/go-multihash"
"golang.org/x/xerrors"
cbg "github.com/whyrusleeping/cbor-gen"
_ "github.com/influxdata/influxdb1-client"
models "github.com/influxdata/influxdb1-client/models"
client "github.com/influxdata/influxdb1-client/v2"
logging "github.com/ipfs/go-log/v2"
)
var log = logging.Logger("stats")
type PointList struct {
points []models.Point
}
func NewPointList() *PointList {
return &PointList{}
}
func (pl *PointList) AddPoint(p models.Point) {
pl.points = append(pl.points, p)
}
func (pl *PointList) Points() []models.Point {
return pl.points
}
type InfluxWriteQueue struct {
ch chan client.BatchPoints
}
func NewInfluxWriteQueue(ctx context.Context, influx client.Client) *InfluxWriteQueue {
ch := make(chan client.BatchPoints, 128)
maxRetries := 10
go func() {
main:
for {
select {
case <-ctx.Done():
return
case batch := <-ch:
for i := 0; i < maxRetries; i++ {
if err := influx.Write(batch); err != nil {
log.Warnw("Failed to write batch", "error", err)
build.Clock.Sleep(15 * time.Second)
continue
}
continue main
}
log.Error("Dropping batch due to failure to write")
}
}
}()
return &InfluxWriteQueue{
ch: ch,
}
}
func (i *InfluxWriteQueue) AddBatch(bp client.BatchPoints) {
i.ch <- bp
}
func (i *InfluxWriteQueue) Close() {
close(i.ch)
}
func InfluxClient(addr, user, pass string) (client.Client, error) {
return client.NewHTTPClient(client.HTTPConfig{
Addr: addr,
Username: user,
Password: pass,
})
}
func InfluxNewBatch() (client.BatchPoints, error) {
return client.NewBatchPoints(client.BatchPointsConfig{})
}
func NewPoint(name string, value interface{}) models.Point {
pt, _ := models.NewPoint(name, models.Tags{},
map[string]interface{}{"value": value}, build.Clock.Now().UTC())
return pt
}
func NewPointFrom(p models.Point) *client.Point {
return client.NewPointFrom(p)
}
func RecordTipsetPoints(ctx context.Context, api v0api.FullNode, pl *PointList, tipset *types.TipSet) error {
cids := []string{}
for _, cid := range tipset.Cids() {
cids = append(cids, cid.String())
}
p := NewPoint("chain.height", int64(tipset.Height()))
p.AddTag("tipset", strings.Join(cids, " "))
pl.AddPoint(p)
p = NewPoint("chain.block_count", len(cids))
pl.AddPoint(p)
tsTime := time.Unix(int64(tipset.MinTimestamp()), int64(0))
p = NewPoint("chain.blocktime", tsTime.Unix())
pl.AddPoint(p)
totalGasLimit := int64(0)
totalUniqGasLimit := int64(0)
seen := make(map[cid.Cid]struct{})
for _, blockheader := range tipset.Blocks() {
bs, err := blockheader.Serialize()
if err != nil {
return err
}
p := NewPoint("chain.election", blockheader.ElectionProof.WinCount)
p.AddTag("miner", blockheader.Miner.String())
pl.AddPoint(p)
p = NewPoint("chain.blockheader_size", len(bs))
pl.AddPoint(p)
msgs, err := api.ChainGetBlockMessages(ctx, blockheader.Cid())
if err != nil {
return xerrors.Errorf("ChainGetBlockMessages failed: %w", msgs)
}
for _, m := range msgs.BlsMessages {
c := m.Cid()
totalGasLimit += m.GasLimit
if _, ok := seen[c]; !ok {
totalUniqGasLimit += m.GasLimit
seen[c] = struct{}{}
}
}
for _, m := range msgs.SecpkMessages {
c := m.Cid()
totalGasLimit += m.Message.GasLimit
if _, ok := seen[c]; !ok {
totalUniqGasLimit += m.Message.GasLimit
seen[c] = struct{}{}
}
}
}
p = NewPoint("chain.gas_limit_total", totalGasLimit)
pl.AddPoint(p)
p = NewPoint("chain.gas_limit_uniq_total", totalUniqGasLimit)
pl.AddPoint(p)
{
baseFeeIn := tipset.Blocks()[0].ParentBaseFee
newBaseFee := store.ComputeNextBaseFee(baseFeeIn, totalUniqGasLimit, len(tipset.Blocks()), tipset.Height())
baseFeeRat := new(big.Rat).SetFrac(newBaseFee.Int, new(big.Int).SetUint64(build.FilecoinPrecision))
baseFeeFloat, _ := baseFeeRat.Float64()
p = NewPoint("chain.basefee", baseFeeFloat)
pl.AddPoint(p)
baseFeeChange := new(big.Rat).SetFrac(newBaseFee.Int, baseFeeIn.Int)
baseFeeChangeF, _ := baseFeeChange.Float64()
p = NewPoint("chain.basefee_change_log", math.Log(baseFeeChangeF)/math.Log(1.125))
pl.AddPoint(p)
}
{
blks := int64(len(cids))
p = NewPoint("chain.gas_fill_ratio", float64(totalGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
p = NewPoint("chain.gas_capacity_ratio", float64(totalUniqGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
p = NewPoint("chain.gas_waste_ratio", float64(totalGasLimit-totalUniqGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
}
return nil
}
type ApiIpldStore struct {
ctx context.Context
api apiIpldStoreApi
}
type apiIpldStoreApi interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
}
func NewApiIpldStore(ctx context.Context, api apiIpldStoreApi) *ApiIpldStore {
return &ApiIpldStore{ctx, api}
}
func (ht *ApiIpldStore) Context() context.Context {
return ht.ctx
}
func (ht *ApiIpldStore) Get(ctx context.Context, c cid.Cid, out interface{}) error {
raw, err := ht.api.ChainReadObj(ctx, c)
if err != nil {
return err
}
cu, ok := out.(cbg.CBORUnmarshaler)
if ok {
if err := cu.UnmarshalCBOR(bytes.NewReader(raw)); err != nil {
return err
}
return nil
}
return fmt.Errorf("Object does not implement CBORUnmarshaler")
}
func (ht *ApiIpldStore) Put(ctx context.Context, v interface{}) (cid.Cid, error) {
return cid.Undef, fmt.Errorf("Put is not implemented on ApiIpldStore")
}
func RecordTipsetStatePoints(ctx context.Context, api v0api.FullNode, pl *PointList, tipset *types.TipSet) error {
attoFil := types.NewInt(build.FilecoinPrecision).Int
//TODO: StatePledgeCollateral API is not implemented and is commented out - re-enable this block once the API is implemented again.
//pc, err := api.StatePledgeCollateral(ctx, tipset.Key())
//if err != nil {
//return err
//}
//pcFil := new(big.Rat).SetFrac(pc.Int, attoFil)
//pcFilFloat, _ := pcFil.Float64()
//p := NewPoint("chain.pledge_collateral", pcFilFloat)
//pl.AddPoint(p)
netBal, err := api.WalletBalance(ctx, reward.Address)
if err != nil {
return err
}
netBalFil := new(big.Rat).SetFrac(netBal.Int, attoFil)
netBalFilFloat, _ := netBalFil.Float64()
p := NewPoint("network.balance", netBalFilFloat)
pl.AddPoint(p)
totalPower, err := api.StateMinerPower(ctx, address.Address{}, tipset.Key())
if err != nil {
return err
}
// We divide the power into gibibytes because 2^63 bytes is 8 exbibytes which is smaller than the Filecoin Mainnet.
// Dividing by a gibibyte gives us more room to work with. This will allow the dashboard to report network and miner
// sizes up to 8192 yobibytes.
gibi := types.NewInt(1024 * 1024 * 1024)
p = NewPoint("chain.power", types.BigDiv(totalPower.TotalPower.QualityAdjPower, gibi).Int64())
pl.AddPoint(p)
powerActor, err := api.StateGetActor(ctx, power.Address, tipset.Key())
if err != nil {
return err
}
powerActorState, err := power.Load(&ApiIpldStore{ctx, api}, powerActor)
if err != nil {
return err
}
return powerActorState.ForEachClaim(func(addr address.Address, claim power.Claim) error {
// BigCmp returns 0 if values are equal
if types.BigCmp(claim.QualityAdjPower, types.NewInt(0)) == 0 {
return nil
}
p = NewPoint("chain.miner_power", types.BigDiv(claim.QualityAdjPower, gibi).Int64())
p.AddTag("miner", addr.String())
pl.AddPoint(p)
return nil
})
}
type msgTag struct {
actor string
method uint64
exitcode uint8
}
func RecordTipsetMessagesPoints(ctx context.Context, api v0api.FullNode, pl *PointList, tipset *types.TipSet) error {
cids := tipset.Cids()
if len(cids) == 0 {
return fmt.Errorf("no cids in tipset")
}
msgs, err := api.ChainGetParentMessages(ctx, cids[0])
if err != nil {
return err
}
recp, err := api.ChainGetParentReceipts(ctx, cids[0])
if err != nil {
return err
}
msgn := make(map[msgTag][]cid.Cid)
totalGasUsed := int64(0)
for _, r := range recp {
totalGasUsed += r.GasUsed
}
p := NewPoint("chain.gas_used_total", totalGasUsed)
pl.AddPoint(p)
for i, msg := range msgs {
// FIXME: use float so this doesn't overflow
// FIXME: this doesn't work as time points get overridden
p := NewPoint("chain.message_gaspremium", msg.Message.GasPremium.Int64())
pl.AddPoint(p)
p = NewPoint("chain.message_gasfeecap", msg.Message.GasFeeCap.Int64())
pl.AddPoint(p)
bs, err := msg.Message.Serialize()
if err != nil {
return err
}
p = NewPoint("chain.message_size", len(bs))
pl.AddPoint(p)
actor, err := api.StateGetActor(ctx, msg.Message.To, tipset.Key())
if err != nil {
return err
}
dm, err := multihash.Decode(actor.Code.Hash())
if err != nil {
continue
}
tag := msgTag{
actor: string(dm.Digest),
method: uint64(msg.Message.Method),
exitcode: uint8(recp[i].ExitCode),
}
found := false
for _, c := range msgn[tag] {
if c.Equals(msg.Cid) {
found = true
break
}
}
if !found {
msgn[tag] = append(msgn[tag], msg.Cid)
}
}
for t, m := range msgn {
p := NewPoint("chain.message_count", len(m))
p.AddTag("actor", t.actor)
p.AddTag("method", fmt.Sprintf("%d", t.method))
p.AddTag("exitcode", fmt.Sprintf("%d", t.exitcode))
pl.AddPoint(p)
}
return nil
}
func ResetDatabase(influx client.Client, database string) error {
log.Info("Resetting database")
q := client.NewQuery(fmt.Sprintf(`DROP DATABASE "%s"; CREATE DATABASE "%s";`, database, database), "", "")
_, err := influx.Query(q)
return err
}
func GetLastRecordedHeight(influx client.Client, database string) (int64, error) {
log.Info("Retrieving last record height")
q := client.NewQuery(`SELECT "value" FROM "chain.height" ORDER BY time DESC LIMIT 1`, database, "")
res, err := influx.Query(q)
if err != nil {
return 0, err
}
if len(res.Results) == 0 {
return 0, fmt.Errorf("No results found for last recorded height")
}
if len(res.Results[0].Series) == 0 {
return 0, fmt.Errorf("No results found for last recorded height")
}
height, err := (res.Results[0].Series[0].Values[0][1].(json.Number)).Int64()
if err != nil {
return 0, err
}
log.Infow("Last record height", "height", height)
return height, nil
}
tools/stats/metrics/metrics.go
+110
View File
@@ -0,0 +1,110 @@
package metrics
import (
"go.opencensus.io/stats"
"go.opencensus.io/stats/view"
"github.com/filecoin-project/lotus/metrics"
)
var Timer = metrics.Timer
var SinceInMilliseconds = metrics.SinceInMilliseconds
// Distribution
var (
defaultMillisecondsDistribution = view.Distribution(0.01, 0.05, 0.1, 0.3, 0.6, 0.8, 1, 2, 3, 4, 5, 6, 8, 10, 13, 16, 32, 64, 128, 256, 500, 1000, 2000, 3000, 5000, 10000, 20000, 30000, 40000, 50000, 60000)
)
// Global Tags
var ()
// Measures
var (
TipsetCollectionHeight = stats.Int64("tipset_collection/height", "Current Height of the node", stats.UnitDimensionless)
TipsetCollectionHeightExpected = stats.Int64("tipset_collection/height_expected", "Current Height of the node", stats.UnitDimensionless)
TipsetCollectionPoints = stats.Int64("tipset_collection/points", "Counter for total number of points collected", stats.UnitDimensionless)
TipsetCollectionDuration = stats.Float64("tipset_collection/total_ms", "Duration of tipset point collection", stats.UnitMilliseconds)
TipsetCollectionBlockHeaderDuration = stats.Float64("tipset_collection/block_header_ms", "Duration of block header point collection", stats.UnitMilliseconds)
TipsetCollectionMessageDuration = stats.Float64("tipset_collection/message_ms", "Duration of message point collection", stats.UnitMilliseconds)
TipsetCollectionStaterootDuration = stats.Float64("tipset_collection/stateroot_ms", "Duration of stateroot point collection", stats.UnitMilliseconds)
IpldStoreCacheSize = stats.Int64("ipld_store/cache_size", "Initialized size of the object read cache", stats.UnitDimensionless)
IpldStoreCacheLength = stats.Int64("ipld_store/cache_length", "Current length of object read cache", stats.UnitDimensionless)
IpldStoreCacheHit = stats.Int64("ipld_store/cache_hit", "Counter for total cache hits", stats.UnitDimensionless)
IpldStoreCacheMiss = stats.Int64("ipld_store/cache_miss", "Counter for total cache misses", stats.UnitDimensionless)
IpldStoreReadDuration = stats.Float64("ipld_store/read_ms", "Duration of object read request to lotus", stats.UnitMilliseconds)
IpldStoreGetDuration = stats.Float64("ipld_store/get_ms", "Duration of object get from store", stats.UnitMilliseconds)
WriteQueueSize = stats.Int64("write_queue/length", "Current length of the write queue", stats.UnitDimensionless)
)
// Views
var (
TipsetCollectionHeightView = &view.View{
Measure: TipsetCollectionHeight,
Aggregation: view.LastValue(),
}
TipsetCollectionHeightExpectedView = &view.View{
Measure: TipsetCollectionHeightExpected,
Aggregation: view.LastValue(),
}
TipsetCollectionPointsView = &view.View{
Measure: TipsetCollectionPoints,
Aggregation: view.Sum(),
}
TipsetCollectionDurationView = &view.View{
Measure: TipsetCollectionDuration,
Aggregation: defaultMillisecondsDistribution,
}
TipsetCollectionBlockHeaderDurationView = &view.View{
Measure: TipsetCollectionBlockHeaderDuration,
Aggregation: defaultMillisecondsDistribution,
}
TipsetCollectionMessageDurationView = &view.View{
Measure: TipsetCollectionMessageDuration,
Aggregation: defaultMillisecondsDistribution,
}
TipsetCollectionStaterootDurationView = &view.View{
Measure: TipsetCollectionStaterootDuration,
Aggregation: defaultMillisecondsDistribution,
}
IpldStoreCacheSizeView = &view.View{
Measure: IpldStoreCacheSize,
Aggregation: view.LastValue(),
}
IpldStoreCacheLengthView = &view.View{
Measure: IpldStoreCacheLength,
Aggregation: view.LastValue(),
}
IpldStoreCacheHitView = &view.View{
Measure: IpldStoreCacheHit,
Aggregation: view.Count(),
}
IpldStoreCacheMissView = &view.View{
Measure: IpldStoreCacheMiss,
Aggregation: view.Count(),
}
IpldStoreReadDurationView = &view.View{
Measure: IpldStoreReadDuration,
Aggregation: defaultMillisecondsDistribution,
}
IpldStoreGetDurationView = &view.View{
Measure: IpldStoreGetDuration,
Aggregation: defaultMillisecondsDistribution,
}
)
// DefaultViews is an array of OpenCensus views for metric gathering purposes
var DefaultViews = []*view.View{
TipsetCollectionHeightView,
TipsetCollectionHeightExpectedView,
TipsetCollectionPointsView,
TipsetCollectionDurationView,
TipsetCollectionBlockHeaderDurationView,
TipsetCollectionMessageDurationView,
TipsetCollectionStaterootDurationView,
IpldStoreCacheSizeView,
IpldStoreCacheLengthView,
IpldStoreCacheHitView,
IpldStoreCacheMissView,
IpldStoreReadDurationView,
IpldStoreGetDurationView,
}
tools/stats/points/collect.go
+363
View File
@@ -0,0 +1,363 @@
package points
import (
"context"
"fmt"
"math"
"math/big"
"strings"
"time"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors/adt"
"github.com/filecoin-project/lotus/chain/actors/builtin/power"
"github.com/filecoin-project/lotus/chain/actors/builtin/reward"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/tools/stats/influx"
"github.com/filecoin-project/lotus/tools/stats/metrics"
lru "github.com/hashicorp/golang-lru"
client "github.com/influxdata/influxdb1-client/v2"
"github.com/ipfs/go-cid"
"github.com/multiformats/go-multihash"
"go.opencensus.io/stats"
"golang.org/x/xerrors"
)
type LotusApi interface {
WalletBalance(context.Context, address.Address) (types.BigInt, error)
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
ChainGetParentMessages(ctx context.Context, blockCid cid.Cid) ([]api.Message, error)
ChainGetParentReceipts(ctx context.Context, blockCid cid.Cid) ([]*types.MessageReceipt, error)
ChainGetBlockMessages(ctx context.Context, blockCid cid.Cid) (*api.BlockMessages, error)
}
type ChainPointCollector struct {
ctx context.Context
api LotusApi
store adt.Store
actorDigestCache *lru.TwoQueueCache
}
func NewChainPointCollector(ctx context.Context, store adt.Store, api LotusApi) (*ChainPointCollector, error) {
actorDigestCache, err := lru.New2Q(2 << 15)
if err != nil {
return nil, err
}
collector := &ChainPointCollector{
ctx: ctx,
store: store,
actorDigestCache: actorDigestCache,
api: api,
}
return collector, nil
}
func (c *ChainPointCollector) actorDigest(ctx context.Context, addr address.Address, tipset *types.TipSet) (string, error) {
if code, ok := c.actorDigestCache.Get(addr); ok {
return code.(string), nil
}
actor, err := c.api.StateGetActor(ctx, addr, tipset.Key())
if err != nil {
return "", err
}
dm, err := multihash.Decode(actor.Code.Hash())
if err != nil {
return "", err
}
digest := string(dm.Digest)
c.actorDigestCache.Add(addr, digest)
return digest, nil
}
func (c *ChainPointCollector) Collect(ctx context.Context, tipset *types.TipSet) (client.BatchPoints, error) {
start := time.Now()
done := metrics.Timer(ctx, metrics.TipsetCollectionDuration)
defer func() {
log.Infow("record tipset", "elapsed", time.Now().Sub(start).Seconds())
done()
}()
pl := influx.NewPointList()
height := tipset.Height()
log.Debugw("collecting tipset points", "height", tipset.Height())
stats.Record(ctx, metrics.TipsetCollectionHeight.M(int64(height)))
if err := c.collectBlockheaderPoints(ctx, pl, tipset); err != nil {
log.Errorw("failed to record tipset", "height", height, "error", err, "tipset", tipset.Key())
}
if err := c.collectMessagePoints(ctx, pl, tipset); err != nil {
log.Errorw("failed to record messages", "height", height, "error", err, "tipset", tipset.Key())
}
if err := c.collectStaterootPoints(ctx, pl, tipset); err != nil {
log.Errorw("failed to record state", "height", height, "error", err, "tipset", tipset.Key())
}
tsTimestamp := time.Unix(int64(tipset.MinTimestamp()), int64(0))
nb, err := influx.NewBatch()
if err != nil {
return nil, err
}
for _, pt := range pl.Points() {
pt.SetTime(tsTimestamp)
nb.AddPoint(influx.NewPointFrom(pt))
}
log.Infow("collected tipset points", "count", len(nb.Points()), "height", tipset.Height())
stats.Record(ctx, metrics.TipsetCollectionPoints.M(int64(len(nb.Points()))))
return nb, nil
}
func (c *ChainPointCollector) collectBlockheaderPoints(ctx context.Context, pl *influx.PointList, tipset *types.TipSet) error {
start := time.Now()
done := metrics.Timer(ctx, metrics.TipsetCollectionBlockHeaderDuration)
defer func() {
log.Infow("collect blockheader points", "elapsed", time.Now().Sub(start).Seconds())
done()
}()
cids := []string{}
for _, cid := range tipset.Cids() {
cids = append(cids, cid.String())
}
p := influx.NewPoint("chain.height", int64(tipset.Height()))
p.AddTag("tipset", strings.Join(cids, " "))
pl.AddPoint(p)
p = influx.NewPoint("chain.block_count", len(cids))
pl.AddPoint(p)
tsTime := time.Unix(int64(tipset.MinTimestamp()), int64(0))
p = influx.NewPoint("chain.blocktime", tsTime.Unix())
pl.AddPoint(p)
totalGasLimit := int64(0)
totalUniqGasLimit := int64(0)
seen := make(map[cid.Cid]struct{})
for _, blockheader := range tipset.Blocks() {
bs, err := blockheader.Serialize()
if err != nil {
return err
}
p := influx.NewPoint("chain.election", blockheader.ElectionProof.WinCount)
p.AddTag("miner", blockheader.Miner.String())
pl.AddPoint(p)
p = influx.NewPoint("chain.blockheader_size", len(bs))
pl.AddPoint(p)
msgs, err := c.api.ChainGetBlockMessages(ctx, blockheader.Cid())
if err != nil {
return xerrors.Errorf("ChainGetBlockMessages failed: %w", msgs)
}
for _, m := range msgs.BlsMessages {
c := m.Cid()
totalGasLimit += m.GasLimit
if _, ok := seen[c]; !ok {
totalUniqGasLimit += m.GasLimit
seen[c] = struct{}{}
}
}
for _, m := range msgs.SecpkMessages {
c := m.Cid()
totalGasLimit += m.Message.GasLimit
if _, ok := seen[c]; !ok {
totalUniqGasLimit += m.Message.GasLimit
seen[c] = struct{}{}
}
}
}
p = influx.NewPoint("chain.gas_limit_total", totalGasLimit)
pl.AddPoint(p)
p = influx.NewPoint("chain.gas_limit_uniq_total", totalUniqGasLimit)
pl.AddPoint(p)
{
baseFeeIn := tipset.Blocks()[0].ParentBaseFee
newBaseFee := store.ComputeNextBaseFee(baseFeeIn, totalUniqGasLimit, len(tipset.Blocks()), tipset.Height())
baseFeeRat := new(big.Rat).SetFrac(newBaseFee.Int, new(big.Int).SetUint64(build.FilecoinPrecision))
baseFeeFloat, _ := baseFeeRat.Float64()
p = influx.NewPoint("chain.basefee", baseFeeFloat)
pl.AddPoint(p)
baseFeeChange := new(big.Rat).SetFrac(newBaseFee.Int, baseFeeIn.Int)
baseFeeChangeF, _ := baseFeeChange.Float64()
p = influx.NewPoint("chain.basefee_change_log", math.Log(baseFeeChangeF)/math.Log(1.125))
pl.AddPoint(p)
}
{
blks := int64(len(cids))
p = influx.NewPoint("chain.gas_fill_ratio", float64(totalGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
p = influx.NewPoint("chain.gas_capacity_ratio", float64(totalUniqGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
p = influx.NewPoint("chain.gas_waste_ratio", float64(totalGasLimit-totalUniqGasLimit)/float64(blks*build.BlockGasTarget))
pl.AddPoint(p)
}
return nil
}
func (c *ChainPointCollector) collectStaterootPoints(ctx context.Context, pl *influx.PointList, tipset *types.TipSet) error {
start := time.Now()
done := metrics.Timer(ctx, metrics.TipsetCollectionStaterootDuration)
defer func() {
log.Infow("collect stateroot points", "elapsed", time.Now().Sub(start).Seconds())
done()
}()
attoFil := types.NewInt(build.FilecoinPrecision).Int
netBal, err := c.api.WalletBalance(ctx, reward.Address)
if err != nil {
return err
}
netBalFil := new(big.Rat).SetFrac(netBal.Int, attoFil)
netBalFilFloat, _ := netBalFil.Float64()
p := influx.NewPoint("network.balance", netBalFilFloat)
pl.AddPoint(p)
totalPower, err := c.api.StateMinerPower(ctx, address.Address{}, tipset.Key())
if err != nil {
return err
}
// We divide the power into gibibytes because 2^63 bytes is 8 exbibytes which is smaller than the Filecoin Mainnet.
// Dividing by a gibibyte gives us more room to work with. This will allow the dashboard to report network and miner
// sizes up to 8192 yobibytes.
gibi := types.NewInt(1024 * 1024 * 1024)
p = influx.NewPoint("chain.power", types.BigDiv(totalPower.TotalPower.QualityAdjPower, gibi).Int64())
pl.AddPoint(p)
powerActor, err := c.api.StateGetActor(ctx, power.Address, tipset.Key())
if err != nil {
return err
}
powerActorState, err := power.Load(c.store, powerActor)
if err != nil {
return err
}
return powerActorState.ForEachClaim(func(addr address.Address, claim power.Claim) error {
// BigCmp returns 0 if values are equal
if types.BigCmp(claim.QualityAdjPower, types.NewInt(0)) == 0 {
return nil
}
p = influx.NewPoint("chain.miner_power", types.BigDiv(claim.QualityAdjPower, gibi).Int64())
p.AddTag("miner", addr.String())
pl.AddPoint(p)
return nil
})
}
type msgTag struct {
actor string
method uint64
exitcode uint8
}
func (c *ChainPointCollector) collectMessagePoints(ctx context.Context, pl *influx.PointList, tipset *types.TipSet) error {
start := time.Now()
done := metrics.Timer(ctx, metrics.TipsetCollectionMessageDuration)
defer func() {
log.Infow("collect message points", "elapsed", time.Now().Sub(start).Seconds())
done()
}()
cids := tipset.Cids()
if len(cids) == 0 {
return fmt.Errorf("no cids in tipset")
}
msgs, err := c.api.ChainGetParentMessages(ctx, cids[0])
if err != nil {
return err
}
recp, err := c.api.ChainGetParentReceipts(ctx, cids[0])
if err != nil {
return err
}
msgn := make(map[msgTag][]cid.Cid)
totalGasUsed := int64(0)
for _, r := range recp {
totalGasUsed += r.GasUsed
}
p := influx.NewPoint("chain.gas_used_total", totalGasUsed)
pl.AddPoint(p)
for i, msg := range msgs {
digest, err := c.actorDigest(ctx, msg.Message.To, tipset)
if err != nil {
continue
}
// FIXME: use float so this doesn't overflow
// FIXME: this doesn't work as time points get overridden
p := influx.NewPoint("chain.message_gaspremium", msg.Message.GasPremium.Int64())
pl.AddPoint(p)
p = influx.NewPoint("chain.message_gasfeecap", msg.Message.GasFeeCap.Int64())
pl.AddPoint(p)
bs, err := msg.Message.Serialize()
if err != nil {
return err
}
p = influx.NewPoint("chain.message_size", len(bs))
pl.AddPoint(p)
tag := msgTag{
actor: digest,
method: uint64(msg.Message.Method),
exitcode: uint8(recp[i].ExitCode),
}
found := false
for _, c := range msgn[tag] {
if c.Equals(msg.Cid) {
found = true
break
}
}
if !found {
msgn[tag] = append(msgn[tag], msg.Cid)
}
}
for t, m := range msgn {
p := influx.NewPoint("chain.message_count", len(m))
p.AddTag("actor", t.actor)
p.AddTag("method", fmt.Sprintf("%d", t.method))
p.AddTag("exitcode", fmt.Sprintf("%d", t.exitcode))
pl.AddPoint(p)
}
return nil
}
tools/stats/points/log.go
+7
View File
@@ -0,0 +1,7 @@
package points
import (
logging "github.com/ipfs/go-log/v2"
)
var log = logging.Logger("stats/points")
tools/stats/rpc.go
-228
View File
@@ -1,228 +0,0 @@
package stats
import (
"context"
"net/http"
"time"
"github.com/filecoin-project/go-jsonrpc"
"github.com/filecoin-project/go-state-types/abi"
manet "github.com/multiformats/go-multiaddr/net"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/client"
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/repo"
)
func getAPI(path string) (string, http.Header, error) {
r, err := repo.NewFS(path)
if err != nil {
return "", nil, err
}
ma, err := r.APIEndpoint()
if err != nil {
return "", nil, xerrors.Errorf("failed to get api endpoint: %w", err)
}
_, addr, err := manet.DialArgs(ma)
if err != nil {
return "", nil, err
}
var headers http.Header
token, err := r.APIToken()
if err != nil {
log.Warnw("Couldn't load CLI token, capabilities may be limited", "error", err)
} else {
headers = http.Header{}
headers.Add("Authorization", "Bearer "+string(token))
}
return "ws://" + addr + "/rpc/v0", headers, nil
}
func WaitForSyncComplete(ctx context.Context, napi v0api.FullNode) error {
sync_complete:
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-build.Clock.After(5 * time.Second):
state, err := napi.SyncState(ctx)
if err != nil {
return err
}
for i, w := range state.ActiveSyncs {
if w.Target == nil {
continue
}
if w.Stage == api.StageSyncErrored {
log.Errorw(
"Syncing",
"worker", i,
"base", w.Base.Key(),
"target", w.Target.Key(),
"target_height", w.Target.Height(),
"height", w.Height,
"error", w.Message,
"stage", w.Stage.String(),
)
} else {
log.Infow(
"Syncing",
"worker", i,
"base", w.Base.Key(),
"target", w.Target.Key(),
"target_height", w.Target.Height(),
"height", w.Height,
"stage", w.Stage.String(),
)
}
if w.Stage == api.StageSyncComplete {
break sync_complete
}
}
}
}
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-build.Clock.After(5 * time.Second):
head, err := napi.ChainHead(ctx)
if err != nil {
return err
}
timestampDelta := build.Clock.Now().Unix() - int64(head.MinTimestamp())
log.Infow(
"Waiting for reasonable head height",
"height", head.Height(),
"timestamp_delta", timestampDelta,
)
// If we get within 20 blocks of the current exected block height we
// consider sync complete. Block propagation is not always great but we still
// want to be recording stats as soon as we can
if timestampDelta < int64(build.BlockDelaySecs)*20 {
return nil
}
}
}
}
func GetTips(ctx context.Context, api v0api.FullNode, lastHeight abi.ChainEpoch, headlag int) (<-chan *types.TipSet, error) {
chmain := make(chan *types.TipSet)
hb := newHeadBuffer(headlag)
notif, err := api.ChainNotify(ctx)
if err != nil {
return nil, err
}
go func() {
defer close(chmain)
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case changes, ok := <-notif:
if !ok {
return
}
for _, change := range changes {
log.Infow("Head event", "height", change.Val.Height(), "type", change.Type)
switch change.Type {
case store.HCCurrent:
tipsets, err := loadTipsets(ctx, api, change.Val, lastHeight)
if err != nil {
log.Info(err)
return
}
for _, tipset := range tipsets {
chmain <- tipset
}
case store.HCApply:
if out := hb.push(change); out != nil {
chmain <- out.Val
}
case store.HCRevert:
hb.pop()
}
}
case <-ticker.C:
log.Info("Running health check")
cctx, cancel := context.WithTimeout(ctx, 5*time.Second)
if _, err := api.ID(cctx); err != nil {
log.Error("Health check failed")
cancel()
return
}
cancel()
log.Info("Node online")
case <-ctx.Done():
return
}
}
}()
return chmain, nil
}
func loadTipsets(ctx context.Context, api v0api.FullNode, curr *types.TipSet, lowestHeight abi.ChainEpoch) ([]*types.TipSet, error) {
tipsets := []*types.TipSet{}
for {
if curr.Height() == 0 {
break
}
if curr.Height() <= lowestHeight {
break
}
log.Infow("Walking back", "height", curr.Height())
tipsets = append(tipsets, curr)
tsk := curr.Parents()
prev, err := api.ChainGetTipSet(ctx, tsk)
if err != nil {
return tipsets, err
}
curr = prev
}
for i, j := 0, len(tipsets)-1; i < j; i, j = i+1, j-1 {
tipsets[i], tipsets[j] = tipsets[j], tipsets[i]
}
return tipsets, nil
}
func GetFullNodeAPI(ctx context.Context, repo string) (v0api.FullNode, jsonrpc.ClientCloser, error) {
addr, headers, err := getAPI(repo)
if err != nil {
return nil, nil, err
}
return client.NewFullNodeRPCV0(ctx, addr, headers)
}
tools/stats/sync/log.go
+7
View File
@@ -0,0 +1,7 @@
package sync
import (
logging "github.com/ipfs/go-log/v2"
)
var log = logging.Logger("stats/sync")
tools/stats/sync/sync.go
+192
View File
@@ -0,0 +1,192 @@
package sync
import (
"context"
"time"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/tools/stats/headbuffer"
)
type SyncWaitApi interface {
SyncState(context.Context) (*api.SyncState, error)
ChainHead(context.Context) (*types.TipSet, error)
}
// SyncWait returns when ChainHead is within 20 epochs of the expected height
func SyncWait(ctx context.Context, napi SyncWaitApi) error {
for {
state, err := napi.SyncState(ctx)
if err != nil {
return err
}
if len(state.ActiveSyncs) == 0 {
build.Clock.Sleep(time.Second)
continue
}
head, err := napi.ChainHead(ctx)
if err != nil {
return err
}
working := -1
for i, ss := range state.ActiveSyncs {
switch ss.Stage {
case api.StageSyncComplete:
default:
working = i
case api.StageIdle:
// not complete, not actively working
}
}
if working == -1 {
working = len(state.ActiveSyncs) - 1
}
ss := state.ActiveSyncs[working]
if ss.Base == nil || ss.Target == nil {
log.Infow(
"syncing",
"height", ss.Height,
"stage", ss.Stage.String(),
)
} else {
log.Infow(
"syncing",
"base", ss.Base.Key(),
"target", ss.Target.Key(),
"target_height", ss.Target.Height(),
"height", ss.Height,
"stage", ss.Stage.String(),
)
}
if build.Clock.Now().Unix()-int64(head.MinTimestamp()) < int64(build.BlockDelaySecs)*30 {
break
}
select {
case <-ctx.Done():
return ctx.Err()
case <-build.Clock.After(time.Duration(int64(build.BlockDelaySecs) * int64(time.Second))):
}
}
return nil
}
type BufferedTipsetChannelApi interface {
ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
Version(context.Context) (api.APIVersion, error)
ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
}
// BufferedTipsetChannel returns an unbuffered channel of tipsets. Buffering occurs internally to handle revert
// ChainNotify changes. The returned channel can output tipsets at the same height twice if a reorg larger the the
// provided `size` occurs.
func BufferedTipsetChannel(ctx context.Context, api BufferedTipsetChannelApi, lastHeight abi.ChainEpoch, size int) (<-chan *types.TipSet, error) {
chmain := make(chan *types.TipSet)
hb := headbuffer.NewHeadChangeStackBuffer(size)
notif, err := api.ChainNotify(ctx)
if err != nil {
return nil, err
}
go func() {
defer close(chmain)
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case changes, ok := <-notif:
if !ok {
return
}
for _, change := range changes {
log.Debugw("head event", "height", change.Val.Height(), "type", change.Type)
switch change.Type {
case store.HCCurrent:
tipsets, err := loadTipsets(ctx, api, change.Val, lastHeight)
if err != nil {
log.Info(err)
return
}
for _, tipset := range tipsets {
chmain <- tipset
}
case store.HCApply:
if out := hb.Push(change); out != nil {
chmain <- out.Val
}
case store.HCRevert:
hb.Pop()
}
}
case <-ticker.C:
log.Debug("running health check")
cctx, cancel := context.WithTimeout(ctx, 5*time.Second)
if _, err := api.Version(cctx); err != nil {
log.Error("health check failed")
cancel()
return
}
cancel()
log.Debug("node online")
case <-ctx.Done():
return
}
}
}()
return chmain, nil
}
func loadTipsets(ctx context.Context, api BufferedTipsetChannelApi, curr *types.TipSet, lowestHeight abi.ChainEpoch) ([]*types.TipSet, error) {
log.Infow("loading tipsets", "to_height", lowestHeight, "from_height", curr.Height())
tipsets := []*types.TipSet{}
for {
if curr.Height() == 0 {
break
}
if curr.Height() <= lowestHeight {
break
}
log.Debugw("walking back", "height", curr.Height())
tipsets = append(tipsets, curr)
tsk := curr.Parents()
prev, err := api.ChainGetTipSet(ctx, tsk)
if err != nil {
return tipsets, err
}
curr = prev
}
for i, j := 0, len(tipsets)-1; i < j; i, j = i+1, j-1 {
tipsets[i], tipsets[j] = tipsets[j], tipsets[i]
}
return tipsets, nil
}