lotus/chain/metrics/consensus.go

package metrics

import (
	"context"
	"encoding/json"

	"github.com/filecoin-project/specs-actors/actors/abi"
	"github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log/v2"
	pubsub "github.com/libp2p/go-libp2p-pubsub"
	"go.uber.org/fx"

	"github.com/filecoin-project/lotus/build"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/node/impl/full"
	"github.com/filecoin-project/lotus/node/modules/helpers"
)

var log = logging.Logger("metrics")

const baseTopic = "/fil/headnotifs/"

type Update struct {
	Type string
}

func SendHeadNotifs(nickname string) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, chain full.ChainAPI) error {
	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, chain full.ChainAPI) error {
		ctx := helpers.LifecycleCtx(mctx, lc)

		lc.Append(fx.Hook{
			OnStart: func(_ context.Context) error {
				gen, err := chain.Chain.GetGenesis()
				if err != nil {
					return err
				}

				topic := baseTopic + gen.Cid().String()

				go func() {
					if err := sendHeadNotifs(ctx, ps, topic, chain, nickname); err != nil {
						log.Error("consensus metrics error", err)
						return
					}
				}()
				go func() {
					sub, err := ps.Subscribe(topic)
					if err != nil {
						return
					}
					defer sub.Cancel()

					for {
						if _, err := sub.Next(ctx); err != nil {
							return
						}
					}

				}()
				return nil
			},
		})

		return nil
	}
}

type message struct {
	// TipSet
	Cids   []cid.Cid
	Blocks []*types.BlockHeader
	Height abi.ChainEpoch
	Weight types.BigInt
	Time   uint64
	Nonce  uint64

	// Meta

	NodeName string
}

func sendHeadNotifs(ctx context.Context, ps *pubsub.PubSub, topic string, chain full.ChainAPI, nickname string) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	notifs, err := chain.ChainNotify(ctx)
	if err != nil {
		return err
	}

	// using unix nano time makes very sure we pick a nonce higher than previous restart
	nonce := uint64(build.Clock.Now().UnixNano())

	for {
		select {
		case notif := <-notifs:
			n := notif[len(notif)-1]

			w, err := chain.ChainTipSetWeight(ctx, n.Val.Key())
			if err != nil {
				return err
			}

			m := message{
				Cids:     n.Val.Cids(),
				Blocks:   n.Val.Blocks(),
				Height:   n.Val.Height(),
				Weight:   w,
				NodeName: nickname,
				Time:     uint64(build.Clock.Now().UnixNano() / 1000_000),
				Nonce:    nonce,
			}

			b, err := json.Marshal(m)
			if err != nil {
				return err
			}

			if err := ps.Publish(topic, b); err != nil {
				return err
			}
		case <-ctx.Done():
			return nil
		}

		nonce++
	}
}