Merge pull request #1294 from natewalck/more-opencensus-metrics

More opencensus metrics

api/api_full.go

@@ -71,6 +71,7 @@ type FullNode interface {
 	WalletBalance(context.Context, address.Address) (types.BigInt, error)
 	WalletSign(context.Context, address.Address, []byte) (*types.Signature, error)
 	WalletSignMessage(context.Context, address.Address, *types.Message) (*types.SignedMessage, error)
+	WalletVerify(context.Context, address.Address, []byte, *types.Signature) bool
 	WalletDefaultAddress(context.Context) (address.Address, error)
 	WalletSetDefault(context.Context, address.Address) error
 	WalletExport(context.Context, address.Address) (*types.KeyInfo, error)

api/apistruct/struct.go

@@ -83,6 +83,7 @@ type FullNodeStruct struct {
 		WalletBalance        func(context.Context, address.Address) (types.BigInt, error)                         `perm:"read"`
 		WalletSign           func(context.Context, address.Address, []byte) (*types.Signature, error)             `perm:"sign"`
 		WalletSignMessage    func(context.Context, address.Address, *types.Message) (*types.SignedMessage, error) `perm:"sign"`
+		WalletVerify         func(context.Context, address.Address, []byte, *types.Signature) bool                `perm:"read"`
 		WalletDefaultAddress func(context.Context) (address.Address, error)                                       `perm:"write"`
 		WalletSetDefault     func(context.Context, address.Address) error                                         `perm:"admin"`
 		WalletExport         func(context.Context, address.Address) (*types.KeyInfo, error)                       `perm:"admin"`
@@ -311,6 +312,10 @@ func (c *FullNodeStruct) WalletSignMessage(ctx context.Context, k address.Addres
 	return c.Internal.WalletSignMessage(ctx, k, msg)
 }
 
+func (c *FullNodeStruct) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *types.Signature) bool {
+	return c.Internal.WalletVerify(ctx, k, msg, sig)
+}
+
 func (c *FullNodeStruct) WalletDefaultAddress(ctx context.Context) (address.Address, error) {
 	return c.Internal.WalletDefaultAddress(ctx)
 }

chain/store/store.go

@@ -13,6 +13,8 @@ import (
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/state"
 	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/metrics"
+	"go.opencensus.io/stats"
 	"go.opencensus.io/trace"
 	"go.uber.org/multierr"
 
@@ -100,7 +102,15 @@ func NewChainStore(bs bstore.Blockstore, ds dstore.Batching, vmcalls *types.VMSy
 		return nil
 	}
 
-	cs.headChangeNotifs = append(cs.headChangeNotifs, hcnf)
+	hcmetric := func(rev, app []*types.TipSet) error {
+		ctx := context.Background()
+		for _, r := range app {
+			stats.Record(ctx, metrics.ChainNodeHeight.M(int64(r.Height())))
+		}
+		return nil
+	}
+
+	cs.headChangeNotifs = append(cs.headChangeNotifs, hcnf, hcmetric)
 
 	return cs
 }

chain/sub/incoming.go

@@ -2,6 +2,7 @@ package sub
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	lru "github.com/hashicorp/golang-lru"
@@ -10,11 +11,14 @@ import (
 	connmgr "github.com/libp2p/go-libp2p-core/connmgr"
 	peer "github.com/libp2p/go-libp2p-peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
 
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/metrics"
 )
 
 var log = logging.Logger("sub")
@@ -107,15 +111,25 @@ func (bv *BlockValidator) flagPeer(p peer.ID) {
 }
 
 func (bv *BlockValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) bool {
+	stats.Record(ctx, metrics.BlockReceived.M(1))
+	ctx, _ = tag.New(
+		ctx,
+		tag.Insert(metrics.PeerID, pid.String()),
+		tag.Insert(metrics.ReceivedFrom, msg.ReceivedFrom.String()),
+	)
 	blk, err := types.DecodeBlockMsg(msg.GetData())
 	if err != nil {
 		log.Error("got invalid block over pubsub: ", err)
+		ctx, _ = tag.New(ctx, tag.Insert(metrics.FailureType, "invalid"))
+		stats.Record(ctx, metrics.BlockValidationFailure.M(1))
 		bv.flagPeer(pid)
 		return false
 	}
 
 	if len(blk.BlsMessages)+len(blk.SecpkMessages) > build.BlockMessageLimit {
 		log.Warnf("received block with too many messages over pubsub")
+		ctx, _ = tag.New(ctx, tag.Insert(metrics.FailureType, "too_many_messages"))
+		stats.Record(ctx, metrics.BlockValidationFailure.M(1))
 		bv.flagPeer(pid)
 		return false
 	}
@@ -127,6 +141,7 @@ func (bv *BlockValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub
 	}
 
 	msg.ValidatorData = blk
+	stats.Record(ctx, metrics.BlockValidationSuccess.M(1))
 	return true
 }
 
@@ -153,9 +168,44 @@ func (brc *blockReceiptCache) add(bcid cid.Cid) int {
 	return val.(int)
 }
 
+type MessageValidator struct {
+	mpool *messagepool.MessagePool
+}
+
+func NewMessageValidator(mp *messagepool.MessagePool) *MessageValidator {
+	return &MessageValidator{mp}
+}
+
+func (mv *MessageValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) bool {
+	stats.Record(ctx, metrics.MessageReceived.M(1))
+	ctx, _ = tag.New(ctx, tag.Insert(metrics.PeerID, pid.String()))
+	m, err := types.DecodeSignedMessage(msg.Message.GetData())
+	if err != nil {
+		log.Warnf("failed to decode incoming message: %s", err)
+		ctx, _ = tag.New(ctx, tag.Insert(metrics.FailureType, "decode"))
+		stats.Record(ctx, metrics.MessageValidationFailure.M(1))
+		return false
+	}
+
+	if err := mv.mpool.Add(m); err != nil {
+		log.Warnf("failed to add message from network to message pool (From: %s, To: %s, Nonce: %d, Value: %s): %s", m.Message.From, m.Message.To, m.Message.Nonce, types.FIL(m.Message.Value), err)
+		ctx, _ = tag.New(
+			ctx,
+			tag.Insert(metrics.MessageFrom, m.Message.From.String()),
+			tag.Insert(metrics.MessageTo, m.Message.To.String()),
+			tag.Insert(metrics.MessageNonce, fmt.Sprint(m.Message.Nonce)),
+			tag.Insert(metrics.FailureType, "add"),
+		)
+		stats.Record(ctx, metrics.MessageValidationFailure.M(1))
+		return false
+	}
+	stats.Record(ctx, metrics.MessageValidationSuccess.M(1))
+	return true
+}
+
 func HandleIncomingMessages(ctx context.Context, mpool *messagepool.MessagePool, msub *pubsub.Subscription) {
 	for {
-		msg, err := msub.Next(ctx)
+		_, err := msub.Next(ctx)
 		if err != nil {
 			log.Warn("error from message subscription: ", err)
 			if ctx.Err() != nil {
@@ -165,15 +215,6 @@ func HandleIncomingMessages(ctx context.Context, mpool *messagepool.MessagePool,
 			continue
 		}
 
-		m, ok := msg.ValidatorData.(*types.SignedMessage)
-		if !ok {
-			log.Errorf("message validator func passed on wrong type: %#v", msg.ValidatorData)
-			continue
-		}
-
-		if err := mpool.Add(m); err != nil {
-			log.Warnf("failed to add message from network to message pool (From: %s, To: %s, Nonce: %d, Value: %s): %s", m.Message.From, m.Message.To, m.Message.Nonce, types.FIL(m.Message.Value), err)
-			continue
-		}
+		// Do nothing... everything happens in validate
 	}
 }

chain/sync.go

@@ -23,6 +23,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"github.com/whyrusleeping/pubsub"
+	"go.opencensus.io/stats"
 	"go.opencensus.io/trace"
 	"golang.org/x/xerrors"
 
@@ -38,6 +39,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/lib/sigs"
+	"github.com/filecoin-project/lotus/metrics"
 )
 
 var log = logging.Logger("chain")
@@ -1038,6 +1040,7 @@ func (syncer *Syncer) syncMessagesAndCheckState(ctx context.Context, headers []*
 			return xerrors.Errorf("message processing failed: %w", err)
 		}
 
+		stats.Record(ctx, metrics.ChainNodeWorkerHeight.M(int64(fts.TipSet().Height())))
 		ss.SetHeight(fts.TipSet().Height())
 
 		return nil

cli/cmd.go

@@ -28,6 +28,20 @@ const (
 	metadataTraceConetxt = "traceContext"
 )
 
+// custom CLI error
+
+type ErrCmdFailed struct {
+	msg string
+}
+
+func (e *ErrCmdFailed) Error() string {
+	return e.msg
+}
+
+func NewCliError(s string) error {
+	return &ErrCmdFailed{s}
+}
+
 // ApiConnector returns API instance
 type ApiConnector func() api.FullNode
 

cli/wallet.go

@@ -24,6 +24,8 @@ var walletCmd = &cli.Command{
 		walletImport,
 		walletGetDefault,
 		walletSetDefault,
+		walletSign,
+		walletVerify,
 	},
 }
 
@@ -235,3 +237,94 @@ var walletImport = &cli.Command{
 		return nil
 	},
 }
+
+var walletSign = &cli.Command{
+	Name:      "sign",
+	Usage:     "sign a message",
+	ArgsUsage: "<signing address> <hexMessage>",
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := ReqContext(cctx)
+
+		if !cctx.Args().Present() || cctx.NArg() != 2 {
+			return fmt.Errorf("must specify signing address and message to sign")
+		}
+
+		addr, err := address.NewFromString(cctx.Args().First())
+
+		if err != nil {
+			return err
+		}
+
+		msg, err := hex.DecodeString(cctx.Args().Get(1))
+
+		if err != nil {
+			return err
+		}
+
+		sig, err := api.WalletSign(ctx, addr, msg)
+
+		if err != nil {
+			return err
+		}
+
+		sigBytes := append([]byte{byte(sig.TypeCode())}, sig.Data...)
+
+		fmt.Println(hex.EncodeToString(sigBytes))
+		return nil
+	},
+}
+
+var walletVerify = &cli.Command{
+	Name:      "verify",
+	Usage:     "verify the signature of a message",
+	ArgsUsage: "<signing address> <hexMessage> <signature>",
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := GetFullNodeAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := ReqContext(cctx)
+
+		if !cctx.Args().Present() || cctx.NArg() != 3 {
+			return fmt.Errorf("must specify signing address, message, and signature to verify")
+		}
+
+		addr, err := address.NewFromString(cctx.Args().First())
+
+		if err != nil {
+			return err
+		}
+
+		msg, err := hex.DecodeString(cctx.Args().Get(1))
+
+		if err != nil {
+			return err
+		}
+
+		sigBytes, err := hex.DecodeString(cctx.Args().Get(2))
+
+		if err != nil {
+			return err
+		}
+
+		sig, err := types.SignatureFromBytes(sigBytes)
+
+		if err != nil {
+			return err
+		}
+
+		if api.WalletVerify(ctx, addr, msg, &sig) {
+			fmt.Println("valid")
+			return nil
+		} else {
+			fmt.Println("invalid")
+			return NewCliError("CLI Verify called with invalid signature")
+		}
+	},
+}

cmd/lotus/daemon.go

@@ -24,6 +24,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node"
 	"github.com/filecoin-project/lotus/node/modules"
 	"github.com/filecoin-project/lotus/node/modules/testing"
@@ -36,12 +37,6 @@ const (
 	preSealedSectorsFlag = "genesis-presealed-sectors"
 )
 
-var (
-	lotusInfo  = stats.Int64("info", "Arbitrary counter to tag lotus info to", stats.UnitDimensionless)
-	version, _ = tag.NewKey("version")
-	commit, _  = tag.NewKey("commit")
-)
-
 // DaemonCmd is the `go-lotus daemon` command
 var DaemonCmd = &cli.Command{
 	Name:  "daemon",
@@ -99,7 +94,7 @@ var DaemonCmd = &cli.Command{
 			defer pprof.StopCPUProfile()
 		}
 
-		ctx, _ := tag.New(context.Background(), tag.Insert(version, build.BuildVersion), tag.Insert(commit, build.CurrentCommit))
+		ctx, _ := tag.New(context.Background(), tag.Insert(metrics.Version, build.BuildVersion), tag.Insert(metrics.Commit, build.CurrentCommit))
 		{
 			dir, err := homedir.Expand(cctx.String("repo"))
 			if err != nil {
@@ -180,21 +175,15 @@ var DaemonCmd = &cli.Command{
 			return xerrors.Errorf("initializing node: %w", err)
 		}
 
-		// We are using this metric to tag info about lotus even though
-		// it doesn't contain any actual metrics
+		// Register all metric views
 		if err = view.Register(
-			&view.View{
-				Name:        "info",
-				Description: "Lotus node information",
-				Measure:     lotusInfo,
-				Aggregation: view.LastValue(),
-				TagKeys:     []tag.Key{version, commit},
-			},
+			metrics.DefaultViews...,
 		); err != nil {
 			log.Fatalf("Cannot register the view: %v", err)
 		}
+
 		// Set the metric to one so it is published to the exporter
-		stats.Record(ctx, lotusInfo.M(1))
+		stats.Record(ctx, metrics.LotusInfo.M(1))
 
 		endpoint, err := r.APIEndpoint()
 		if err != nil {

cmd/lotus/main.go

@@ -73,7 +73,12 @@ func main() {
 			Code:    trace.StatusCodeFailedPrecondition,
 			Message: err.Error(),
 		})
-		log.Warnf("%+v", err)
+		_, ok := err.(*lcli.ErrCmdFailed)
+		if ok {
+			log.Debugf("%+v", err)
+		} else {
+			log.Warnf("%+v", err)
+		}
 		os.Exit(1)
 	}
 }

documentation/en/.library.json

@@ -111,6 +111,13 @@
       "value": null,
       "posts": []
     },
+    {
+      "title": "Command Line Interface",
+      "slug": "en+cli",
+      "github": "en/cli.md",
+      "value": null,
+      "posts": []
+    },
     {
       "title": "API",
       "slug": "en+api",

documentation/en/cli.md

@@ -0,0 +1,108 @@
+# Lotus Command Line Interface
+
+The Command Line Interface (CLI) is a convenient way to interact with
+a Lotus node. You can use the CLI to operate your node,
+get information about the blockchain,
+manage your accounts and transfer funds,
+create storage deals, and much more! 
+
+The CLI is intended to be self-documenting, so when in doubt, simply add `--help`
+to whatever command you're trying to run! This will also display all of the 
+input parameters that can be provided to a command.
+
+We highlight some of the commonly
+used features of the CLI below.
+All CLI commands should be run from the home directory of the Lotus project.
+
+## Operating a Lotus node
+
+### Starting up a node
+
+```sh
+lotus daemon
+```
+This command will start up your Lotus node, with its API port open at 1234. 
+You can pass `--api=<number>` to use a different port.
+
+### Checking your sync progress
+
+```sh
+lotus sync status
+```
+This command will print your current tipset height under `Height`, and the target tipset height
+under `Taregt`. 
+
+You can also run `lotus sync wait` to get constant updates on your sync progress.
+
+### Getting the head tipset
+
+```sh
+lotus chain head
+```
+
+### Control the logging level
+
+```sh
+lotus log set-level
+```
+This command can be used to toggle the logging levels of the different
+systems of a Lotus node. In decreasing order
+of logging detail, the levels are `debug`, `info`, `warn`, and `error`. 
+
+As an example,
+to set the `chain` and `blocksync` to log at the `debug` level, run 
+`lotus log set-level --system chain --system blocksync debug`. 
+
+To see the various logging system, run `lotus log list`.
+
+### Find out what version of Lotus you're running
+
+```sh
+lotus version
+```
+
+## Managing your accounts
+
+### Listing accounts in your wallet
+
+```sh
+lo*tus wallet list
+``` 
+
+### Creating a new account
+
+```sh
+lotus wallet new bls
+```
+This command will create a new BLS account in your wallet; these
+addresses start with the prefix `t3`. Running `lotus wallet new secp256k1` 
+(or just `lotus wallet new`) will create
+a new Secp256k1 account, which begins with the prefix `t1`.
+
+### Getting an account's balance
+
+```sh
+lotus wallet balance <address>
+``` 
+
+### Transferring funds 
+
+```sh
+lotus send --source=<source address> <destination address> <amount>
+``` 
+This command will transfer `amount` (in attoFIL) from `source address` to `destination address`.
+
+### Importing an account into your wallet
+
+```sh
+lotus wallet import <path to private key>
+``` 
+This command will import an account whose private key is saved at the specified file.
+
+### Exporting an account from your wallet
+
+```sh
+lotus wallet export <address>
+``` 
+This command will print out the private key of the specified address
+if it is in your wallet. Always be careful with your private key!

documentation/en/join-testnet.md

@@ -36,13 +36,13 @@ In order to connect to the network, you need to be connected to at least 1 peer.
 
 ## Chain sync
 
-While the daemon is running, the next requirement is to sync the chain. Run the command below to start the chain sync progress. To see current chain height, visit the [network stats page](https://stats.testnet.filecoin.io/).
+While the daemon is running, the next requirement is to sync the chain. Run the command below to view the chain sync progress. To see current chain height, visit the [network stats page](https://stats.testnet.filecoin.io/).
 
 ```sh
 lotus sync wait
 ```
 
-- This step will take anywhere between 30 minutes to a few hours.
+- This step will take anywhere between a few hours to a couple of days.
 - You will be able to perform **Lotus Testnet** operations after it is finished.
 
 ## Create your first address

go.mod

@@ -84,7 +84,6 @@ require (
 	github.com/multiformats/go-varint v0.0.2
 	github.com/opentracing/opentracing-go v1.1.0
 	github.com/polydawn/refmt v0.0.0-20190809202753-05966cbd336a
-	github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829
 	github.com/prometheus/common v0.2.0
 	github.com/stretchr/testify v1.4.0
 	github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba

lib/jsonrpc/handler.go

@@ -9,6 +9,9 @@ import (
 	"io"
 	"reflect"
 
+	"github.com/filecoin-project/lotus/metrics"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
 	"go.opencensus.io/trace"
 	"go.opencensus.io/trace/propagation"
 	"golang.org/x/xerrors"
@@ -151,18 +154,22 @@ func (handlers) getSpan(ctx context.Context, req request) (context.Context, *tra
 }
 
 func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer)), rpcError rpcErrFunc, done func(keepCtx bool), chOut chanOut) {
+	// Not sure if we need to sanitize the incoming req.Method or not.
 	ctx, span := h.getSpan(ctx, req)
+	ctx, _ = tag.New(ctx, tag.Insert(metrics.RPCMethod, req.Method))
 	defer span.End()
 
 	handler, ok := h[req.Method]
 	if !ok {
 		rpcError(w, &req, rpcMethodNotFound, fmt.Errorf("method '%s' not found", req.Method))
+		stats.Record(ctx, metrics.RPCInvalidMethod.M(1))
 		done(false)
 		return
 	}
 
 	if len(req.Params) != handler.nParams {
 		rpcError(w, &req, rpcInvalidParams, fmt.Errorf("wrong param count"))
+		stats.Record(ctx, metrics.RPCRequestError.M(1))
 		done(false)
 		return
 	}
@@ -172,6 +179,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 
 	if chOut == nil && outCh {
 		rpcError(w, &req, rpcMethodNotFound, fmt.Errorf("method '%s' not supported in this mode (no out channel support)", req.Method))
+		stats.Record(ctx, metrics.RPCRequestError.M(1))
 		return
 	}
 
@@ -185,6 +193,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 		rp := reflect.New(handler.paramReceivers[i])
 		if err := json.NewDecoder(bytes.NewReader(req.Params[i].data)).Decode(rp.Interface()); err != nil {
 			rpcError(w, &req, rpcParseError, xerrors.Errorf("unmarshaling params for '%s': %w", handler.handlerFunc, err))
+			stats.Record(ctx, metrics.RPCRequestError.M(1))
 			return
 		}
 
@@ -196,6 +205,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 	callResult, err := doCall(req.Method, handler.handlerFunc, callParams)
 	if err != nil {
 		rpcError(w, &req, 0, xerrors.Errorf("fatal error calling '%s': %w", req.Method, err))
+		stats.Record(ctx, metrics.RPCRequestError.M(1))
 		return
 	}
 	if req.ID == nil {
@@ -213,6 +223,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 		err := callResult[handler.errOut].Interface()
 		if err != nil {
 			log.Warnf("error in RPC call to '%s': %+v", req.Method, err)
+			stats.Record(ctx, metrics.RPCResponseError.M(1))
 			resp.Error = &respError{
 				Code:    1,
 				Message: err.(error).Error(),
@@ -234,6 +245,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 		}
 
 		log.Warnf("failed to setup channel in RPC call to '%s': %+v", req.Method, err)
+		stats.Record(ctx, metrics.RPCResponseError.M(1))
 		resp.Error = &respError{
 			Code:    1,
 			Message: err.(error).Error(),
@@ -243,6 +255,7 @@ func (h handlers) handle(ctx context.Context, req request, w func(func(io.Writer
 	w(func(w io.Writer) {
 		if err := json.NewEncoder(w).Encode(resp); err != nil {
 			log.Error(err)
+			stats.Record(ctx, metrics.RPCResponseError.M(1))
 			return
 		}
 	})

metrics/metrics.go

@@ -0,0 +1,102 @@
+package metrics
+
+import (
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+)
+
+// Global Tags
+var (
+	Version, _      = tag.NewKey("version")
+	Commit, _       = tag.NewKey("commit")
+	RPCMethod, _    = tag.NewKey("method")
+	PeerID, _       = tag.NewKey("peer_id")
+	FailureType, _  = tag.NewKey("failure_type")
+	MessageFrom, _  = tag.NewKey("message_from")
+	MessageTo, _    = tag.NewKey("message_to")
+	MessageNonce, _ = tag.NewKey("message_nonce")
+	ReceivedFrom, _ = tag.NewKey("received_from")
+)
+
+// Measures
+var (
+	LotusInfo                = stats.Int64("info", "Arbitrary counter to tag lotus info to", stats.UnitDimensionless)
+	ChainNodeHeight          = stats.Int64("chain/node_height", "Current Height of the node", stats.UnitDimensionless)
+	ChainNodeWorkerHeight    = stats.Int64("chain/node_worker_height", "Current Height of workers on the node", stats.UnitDimensionless)
+	MessageReceived          = stats.Int64("message/received", "Counter for total received messages", stats.UnitDimensionless)
+	MessageValidationFailure = stats.Int64("message/failure", "Counter for message validation failures", stats.UnitDimensionless)
+	MessageValidationSuccess = stats.Int64("message/success", "Counter for message validation successes", stats.UnitDimensionless)
+	BlockReceived            = stats.Int64("block/received", "Counter for total received blocks", stats.UnitDimensionless)
+	BlockValidationFailure   = stats.Int64("block/failure", "Counter for block validation failures", stats.UnitDimensionless)
+	BlockValidationSuccess   = stats.Int64("block/success", "Counter for block validation successes", stats.UnitDimensionless)
+	PeerCount                = stats.Int64("peer/count", "Current number of FIL peers", stats.UnitDimensionless)
+	RPCInvalidMethod         = stats.Int64("rpc/invalid_method", "Total number of invalid RPC methods called", stats.UnitDimensionless)
+	RPCRequestError          = stats.Int64("rpc/request_error", "Total number of request errors handled", stats.UnitDimensionless)
+	RPCResponseError         = stats.Int64("rpc/response_error", "Total number of responses errors handled", stats.UnitDimensionless)
+)
+
+// DefaultViews is an array of Consensus views for metric gathering purposes
+var DefaultViews = []*view.View{
+	&view.View{
+		Name:        "info",
+		Description: "Lotus node information",
+		Measure:     LotusInfo,
+		Aggregation: view.LastValue(),
+		TagKeys:     []tag.Key{Version, Commit},
+	},
+	&view.View{
+		Measure:     ChainNodeHeight,
+		Aggregation: view.LastValue(),
+	},
+	&view.View{
+		Measure:     ChainNodeWorkerHeight,
+		Aggregation: view.LastValue(),
+	},
+	&view.View{
+		Measure:     BlockReceived,
+		Aggregation: view.Count(),
+	},
+	&view.View{
+		Measure:     BlockValidationFailure,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{FailureType, PeerID, ReceivedFrom},
+	},
+	&view.View{
+		Measure:     BlockValidationSuccess,
+		Aggregation: view.Count(),
+	},
+	&view.View{
+		Measure:     MessageReceived,
+		Aggregation: view.Count(),
+	},
+	&view.View{
+		Measure:     MessageValidationFailure,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{FailureType, MessageFrom, MessageTo, MessageNonce},
+	},
+	&view.View{
+		Measure:     MessageValidationSuccess,
+		Aggregation: view.Count(),
+	},
+	&view.View{
+		Measure:     PeerCount,
+		Aggregation: view.LastValue(),
+	},
+	// All RPC related metrics should at the very least tag the RPCMethod
+	&view.View{
+		Measure:     RPCInvalidMethod,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{RPCMethod},
+	},
+	&view.View{
+		Measure:     RPCRequestError,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{RPCMethod},
+	},
+	&view.View{
+		Measure:     RPCResponseError,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{RPCMethod},
+	},
+}

node/config/def.go

@@ -96,7 +96,7 @@ func DefaultStorageMiner() *StorageMiner {
 		Common: defCommon(),
 
 		SectorBuilder: SectorBuilder{
-			WorkerCount: 5,
+			WorkerCount: 2,
 		},
 	}
 	cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"

node/impl/full/wallet.go

@@ -2,6 +2,7 @@ package full
 
 import (
 	"context"
+	"github.com/filecoin-project/lotus/lib/sigs"
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/lotus/chain/stmgr"
@@ -53,6 +54,10 @@ func (a *WalletAPI) WalletSignMessage(ctx context.Context, k address.Address, ms
 	}, nil
 }
 
+func (a *WalletAPI) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *types.Signature) bool {
+	return sigs.Verify(sig, k, msg) == nil
+}
+
 func (a *WalletAPI) WalletDefaultAddress(ctx context.Context) (address.Address, error) {
 	return a.Wallet.GetDefault()
 }

node/modules/services.go

@@ -16,7 +16,6 @@ import (
 	"github.com/filecoin-project/lotus/chain/blocksync"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/sub"
-	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/node/hello"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -78,18 +77,9 @@ func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub
 		panic(err)
 	}
 
-	v := func(ctx context.Context, pid peer.ID, msg *pubsub.Message) bool {
-		m, err := types.DecodeSignedMessage(msg.GetData())
-		if err != nil {
-			log.Errorf("got incorrectly formatted Message: %s", err)
-			return false
-		}
+	v := sub.NewMessageValidator(mpool)
 
-		msg.ValidatorData = m
-		return true
-	}
-
-	if err := ps.RegisterTopicValidator(MessagesTopic, v); err != nil {
+	if err := ps.RegisterTopicValidator(MessagesTopic, v.Validate); err != nil {
 		panic(err)
 	}
 

peermgr/peermgr.go

@@ -5,7 +5,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"go.opencensus.io/stats"
 	"go.uber.org/fx"
 
 	host "github.com/libp2p/go-libp2p-core/host"
@@ -115,6 +117,7 @@ func (pmgr *PeerMgr) Run(ctx context.Context) {
 			} else if pcount > pmgr.maxFilPeers {
 				log.Debug("peer count about threshold: %d > %d", pcount, pmgr.maxFilPeers)
 			}
+			stats.Record(ctx, metrics.PeerCount.M(int64(pmgr.getPeerCount())))
 		}
 	}
 }

tools/stats/head_buffer.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"container/list"
+
+	"github.com/filecoin-project/lotus/chain/store"
+)
+
+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 *store.HeadChange) (rethc *store.HeadChange) {
+	if h.buffer.Len() == h.size {
+		var ok bool
+
+		el := h.buffer.Front()
+		rethc, ok = el.Value.(*store.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

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/filecoin-project/lotus/chain/store"
+	"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(&store.HeadChange{Type: "1"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "2"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "3"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "4"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "5"}))
+
+		hc := hb.Push(&store.HeadChange{Type: "6"})
+		require.Equal(t, hc.Type, "1")
+	})
+
+	t.Run("Reverts", func(t *testing.T) {
+		hb := NewHeadBuffer(5)
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "1"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "2"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "3"}))
+		hb.Pop()
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "3a"}))
+		hb.Pop()
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "3b"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "4"}))
+		require.Nil(t, hb.Push(&store.HeadChange{Type: "5"}))
+
+		hc := hb.Push(&store.HeadChange{Type: "6"})
+		require.Equal(t, hc.Type, "1")
+		hc = hb.Push(&store.HeadChange{Type: "7"})
+		require.Equal(t, hc.Type, "2")
+		hc = hb.Push(&store.HeadChange{Type: "8"})
+		require.Equal(t, hc.Type, "3b")
+	})
+}

tools/stats/main.go

@@ -22,10 +22,12 @@ func main() {
 	var database string = "lotus"
 	var reset bool = false
 	var height int64 = 0
+	var headlag int = 3
 
 	flag.StringVar(&repo, "repo", repo, "lotus repo path")
 	flag.StringVar(&database, "database", database, "influx database")
 	flag.Int64Var(&height, "height", height, "block height to start syncing from (0 will resume)")
+	flag.IntVar(&headlag, "head-lag", headlag, "number of head events to hold to protect against small reorgs")
 	flag.BoolVar(&reset, "reset", reset, "truncate database before starting stats gathering")
 
 	flag.Parse()
@@ -66,7 +68,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	tipsetsCh, err := GetTips(ctx, api, uint64(height))
+	tipsetsCh, err := GetTips(ctx, api, uint64(height), headlag)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -75,6 +77,7 @@ func main() {
 	defer wq.Close()
 
 	for tipset := range tipsetsCh {
+		log.Infow("Collect stats", "height", tipset.Height())
 		pl := NewPointList()
 		height := tipset.Height()
 

tools/stats/rpc.go

@@ -120,9 +120,11 @@ sync_complete:
 	}
 }
 
-func GetTips(ctx context.Context, api api.FullNode, lastHeight uint64) (<-chan *types.TipSet, error) {
+func GetTips(ctx context.Context, api api.FullNode, lastHeight uint64, 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
@@ -151,7 +153,11 @@ func GetTips(ctx context.Context, api api.FullNode, lastHeight uint64) (<-chan *
 							chmain <- tipset
 						}
 					case store.HCApply:
-						chmain <- change.Val
+						if out := hb.Push(change); out != nil {
+							chmain <- out.Val
+						}
+					case store.HCRevert:
+						hb.Pop()
 					}
 				}
 			case <-ping: