Merge pull request #9 from ChainSafe/mpetrun5/lotus-extended-tracer
Lotus extended pubsub tracer
This commit is contained in:
commit
58afade1bb
@ -1,3 +1,4 @@
|
|||||||
|
//go:build !nodaemon
|
||||||
// +build !nodaemon
|
// +build !nodaemon
|
||||||
|
|
||||||
package main
|
package main
|
||||||
@ -153,6 +154,18 @@ var DaemonCmd = &cli.Command{
|
|||||||
Name: "restore-config",
|
Name: "restore-config",
|
||||||
Usage: "config file to use when restoring from backup",
|
Usage: "config file to use when restoring from backup",
|
||||||
},
|
},
|
||||||
|
&cli.StringFlag{
|
||||||
|
Name: "trace-to-json",
|
||||||
|
Usage: "starts tracer and outputs to json file defined with this flag",
|
||||||
|
},
|
||||||
|
&cli.StringFlag{
|
||||||
|
Name: "trace-to-elasticsearch",
|
||||||
|
Usage: "starts tracer and outputs to elasticsearch, flag must contain connection string for elasticsearch",
|
||||||
|
},
|
||||||
|
&cli.StringFlag{
|
||||||
|
Name: "trace-source-auth",
|
||||||
|
Usage: "auth token for trusted source of traces",
|
||||||
|
},
|
||||||
},
|
},
|
||||||
Action: func(cctx *cli.Context) error {
|
Action: func(cctx *cli.Context) error {
|
||||||
isLite := cctx.Bool("lite")
|
isLite := cctx.Bool("lite")
|
||||||
@ -310,6 +323,10 @@ var DaemonCmd = &cli.Command{
|
|||||||
log.Warnf("unable to inject prometheus ipfs/go-metrics exporter; some metrics will be unavailable; err: %s", err)
|
log.Warnf("unable to inject prometheus ipfs/go-metrics exporter; some metrics will be unavailable; err: %s", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
traceToJsonFile := cctx.String("trace-to-json")
|
||||||
|
traceToElasticsearch := cctx.String("trace-to-elasticsearch")
|
||||||
|
traceSourceAuth := cctx.String("trace-source-auth")
|
||||||
|
|
||||||
var api api.FullNode
|
var api api.FullNode
|
||||||
stop, err := node.New(ctx,
|
stop, err := node.New(ctx,
|
||||||
node.FullAPI(&api, node.Lite(isLite)),
|
node.FullAPI(&api, node.Lite(isLite)),
|
||||||
@ -319,6 +336,9 @@ var DaemonCmd = &cli.Command{
|
|||||||
|
|
||||||
node.Override(new(dtypes.Bootstrapper), isBootstrapper),
|
node.Override(new(dtypes.Bootstrapper), isBootstrapper),
|
||||||
node.Override(new(dtypes.ShutdownChan), shutdownChan),
|
node.Override(new(dtypes.ShutdownChan), shutdownChan),
|
||||||
|
node.Override(new(dtypes.JsonTracer), traceToJsonFile),
|
||||||
|
node.Override(new(dtypes.ElasticSearchTracer), traceToElasticsearch),
|
||||||
|
node.Override(new(dtypes.TracerSourceAuth), traceSourceAuth),
|
||||||
|
|
||||||
genesis,
|
genesis,
|
||||||
liteModeDeps,
|
liteModeDeps,
|
||||||
|
1
go.mod
1
go.mod
@ -22,6 +22,7 @@ require (
|
|||||||
github.com/drand/drand v1.2.1
|
github.com/drand/drand v1.2.1
|
||||||
github.com/drand/kyber v1.1.4
|
github.com/drand/kyber v1.1.4
|
||||||
github.com/dustin/go-humanize v1.0.0
|
github.com/dustin/go-humanize v1.0.0
|
||||||
|
github.com/elastic/go-elasticsearch/v7 v7.14.0
|
||||||
github.com/elastic/go-sysinfo v1.3.0
|
github.com/elastic/go-sysinfo v1.3.0
|
||||||
github.com/elastic/gosigar v0.12.0
|
github.com/elastic/gosigar v0.12.0
|
||||||
github.com/etclabscore/go-openrpc-reflect v0.0.36
|
github.com/etclabscore/go-openrpc-reflect v0.0.36
|
||||||
|
2
go.sum
2
go.sum
@ -235,6 +235,8 @@ github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5m
|
|||||||
github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
|
github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
|
||||||
github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
|
github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
|
||||||
github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
|
github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
|
||||||
|
github.com/elastic/go-elasticsearch/v7 v7.14.0 h1:extp3jos/rwJn3J+lgbaGlwAgs0TVsIHme00GyNAyX4=
|
||||||
|
github.com/elastic/go-elasticsearch/v7 v7.14.0/go.mod h1:OJ4wdbtDNk5g503kvlHLyErCgQwwzmDtaFC4XyOxXA4=
|
||||||
github.com/elastic/go-sysinfo v1.3.0 h1:eb2XFGTMlSwG/yyU9Y8jVAYLIzU2sFzWXwo2gmetyrE=
|
github.com/elastic/go-sysinfo v1.3.0 h1:eb2XFGTMlSwG/yyU9Y8jVAYLIzU2sFzWXwo2gmetyrE=
|
||||||
github.com/elastic/go-sysinfo v1.3.0/go.mod h1:i1ZYdU10oLNfRzq4vq62BEwD2fH8KaWh6eh0ikPT9F0=
|
github.com/elastic/go-sysinfo v1.3.0/go.mod h1:i1ZYdU10oLNfRzq4vq62BEwD2fH8KaWh6eh0ikPT9F0=
|
||||||
github.com/elastic/go-windows v1.0.0 h1:qLURgZFkkrYyTTkvYpsZIgf83AUsdIHfvlJaqaZ7aSY=
|
github.com/elastic/go-windows v1.0.0 h1:qLURgZFkkrYyTTkvYpsZIgf83AUsdIHfvlJaqaZ7aSY=
|
||||||
|
@ -525,6 +525,24 @@ Type: Array of multiaddress peerinfo strings, must include peerid (/p2p/12D3K...
|
|||||||
Name: "RemoteTracer",
|
Name: "RemoteTracer",
|
||||||
Type: "string",
|
Type: "string",
|
||||||
|
|
||||||
|
Comment: ``,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
Name: "JsonTracer",
|
||||||
|
Type: "string",
|
||||||
|
|
||||||
|
Comment: ``,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
Name: "ElasticSearchTracer",
|
||||||
|
Type: "string",
|
||||||
|
|
||||||
|
Comment: ``,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
Name: "TracerSourceAuth",
|
||||||
|
Type: "string",
|
||||||
|
|
||||||
Comment: ``,
|
Comment: ``,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
@ -309,6 +309,10 @@ type Pubsub struct {
|
|||||||
DirectPeers []string
|
DirectPeers []string
|
||||||
IPColocationWhitelist []string
|
IPColocationWhitelist []string
|
||||||
RemoteTracer string
|
RemoteTracer string
|
||||||
|
JsonTracer string
|
||||||
|
ElasticSearchTracer string
|
||||||
|
ElasticSearchIndex string
|
||||||
|
TracerSourceAuth string
|
||||||
}
|
}
|
||||||
|
|
||||||
type Chainstore struct {
|
type Chainstore struct {
|
||||||
|
5
node/modules/dtypes/tracer.go
Normal file
5
node/modules/dtypes/tracer.go
Normal file
@ -0,0 +1,5 @@
|
|||||||
|
package dtypes
|
||||||
|
|
||||||
|
type JsonTracer string
|
||||||
|
type ElasticSearchTracer string
|
||||||
|
type TracerSourceAuth string
|
@ -21,6 +21,7 @@ import (
|
|||||||
"github.com/filecoin-project/lotus/node/config"
|
"github.com/filecoin-project/lotus/node/config"
|
||||||
"github.com/filecoin-project/lotus/node/modules/dtypes"
|
"github.com/filecoin-project/lotus/node/modules/dtypes"
|
||||||
"github.com/filecoin-project/lotus/node/modules/helpers"
|
"github.com/filecoin-project/lotus/node/modules/helpers"
|
||||||
|
"github.com/filecoin-project/lotus/node/modules/tracer"
|
||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
@ -49,6 +50,30 @@ func ScoreKeeper() *dtypes.ScoreKeeper {
|
|||||||
return new(dtypes.ScoreKeeper)
|
return new(dtypes.ScoreKeeper)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type PeerScoreTracker interface {
|
||||||
|
UpdatePeerScore(scores map[peer.ID]*pubsub.PeerScoreSnapshot)
|
||||||
|
}
|
||||||
|
|
||||||
|
type peerScoreTracker struct {
|
||||||
|
sk *dtypes.ScoreKeeper
|
||||||
|
lt tracer.LotusTracer
|
||||||
|
}
|
||||||
|
|
||||||
|
func newPeerScoreTracker(lt tracer.LotusTracer, sk *dtypes.ScoreKeeper) PeerScoreTracker {
|
||||||
|
return &peerScoreTracker{
|
||||||
|
sk: sk,
|
||||||
|
lt: lt,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pst *peerScoreTracker) UpdatePeerScore(scores map[peer.ID]*pubsub.PeerScoreSnapshot) {
|
||||||
|
if pst.lt != nil {
|
||||||
|
pst.lt.PeerScores(scores)
|
||||||
|
}
|
||||||
|
|
||||||
|
pst.sk.Update(scores)
|
||||||
|
}
|
||||||
|
|
||||||
type GossipIn struct {
|
type GossipIn struct {
|
||||||
fx.In
|
fx.In
|
||||||
Mctx helpers.MetricsCtx
|
Mctx helpers.MetricsCtx
|
||||||
@ -272,7 +297,6 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
|
|||||||
OpportunisticGraftThreshold: OpportunisticGraftScoreThreshold,
|
OpportunisticGraftThreshold: OpportunisticGraftScoreThreshold,
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
pubsub.WithPeerScoreInspect(in.Sk.Update, 10*time.Second),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// enable Peer eXchange on bootstrappers
|
// enable Peer eXchange on bootstrappers
|
||||||
@ -341,6 +365,27 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
|
|||||||
pubsub.NewAllowlistSubscriptionFilter(allowTopics...),
|
pubsub.NewAllowlistSubscriptionFilter(allowTopics...),
|
||||||
100)))
|
100)))
|
||||||
|
|
||||||
|
var transports []tracer.TracerTransport
|
||||||
|
if in.Cfg.JsonTracer != "" {
|
||||||
|
jsonTransport, err := tracer.NewJsonTracerTransport(in.Cfg.JsonTracer)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
transports = append(transports, jsonTransport)
|
||||||
|
}
|
||||||
|
if in.Cfg.ElasticSearchTracer != "" {
|
||||||
|
elasticSearchTransport, err := tracer.NewElasticSearchTransport(
|
||||||
|
in.Cfg.ElasticSearchTracer,
|
||||||
|
in.Cfg.ElasticSearchIndex,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
transports = append(transports, elasticSearchTransport)
|
||||||
|
}
|
||||||
|
lt := tracer.NewLotusTracer(transports, in.Host.ID(), in.Cfg.TracerSourceAuth)
|
||||||
|
|
||||||
// tracer
|
// tracer
|
||||||
if in.Cfg.RemoteTracer != "" {
|
if in.Cfg.RemoteTracer != "" {
|
||||||
a, err := ma.NewMultiaddr(in.Cfg.RemoteTracer)
|
a, err := ma.NewMultiaddr(in.Cfg.RemoteTracer)
|
||||||
@ -358,12 +403,18 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
trw := newTracerWrapper(tr, build.BlocksTopic(in.Nn))
|
pst := newPeerScoreTracker(lt, in.Sk)
|
||||||
|
trw := newTracerWrapper(tr, lt, build.BlocksTopic(in.Nn))
|
||||||
|
|
||||||
options = append(options, pubsub.WithEventTracer(trw))
|
options = append(options, pubsub.WithEventTracer(trw))
|
||||||
|
options = append(options, pubsub.WithPeerScoreInspect(pst.UpdatePeerScore, 10*time.Second))
|
||||||
} else {
|
} else {
|
||||||
// still instantiate a tracer for collecting metrics
|
// still instantiate a tracer for collecting metrics
|
||||||
trw := newTracerWrapper(nil)
|
trw := newTracerWrapper(nil, lt)
|
||||||
options = append(options, pubsub.WithEventTracer(trw))
|
options = append(options, pubsub.WithEventTracer(trw))
|
||||||
|
|
||||||
|
pst := newPeerScoreTracker(lt, in.Sk)
|
||||||
|
options = append(options, pubsub.WithPeerScoreInspect(pst.UpdatePeerScore, 10*time.Second))
|
||||||
}
|
}
|
||||||
|
|
||||||
return pubsub.NewGossipSub(helpers.LifecycleCtx(in.Mctx, in.Lc), in.Host, options...)
|
return pubsub.NewGossipSub(helpers.LifecycleCtx(in.Mctx, in.Lc), in.Host, options...)
|
||||||
@ -374,7 +425,11 @@ func HashMsgId(m *pubsub_pb.Message) string {
|
|||||||
return string(hash[:])
|
return string(hash[:])
|
||||||
}
|
}
|
||||||
|
|
||||||
func newTracerWrapper(tr pubsub.EventTracer, topics ...string) pubsub.EventTracer {
|
func newTracerWrapper(
|
||||||
|
lp2pTracer pubsub.EventTracer,
|
||||||
|
lotusTracer pubsub.EventTracer,
|
||||||
|
topics ...string,
|
||||||
|
) pubsub.EventTracer {
|
||||||
var topicsMap map[string]struct{}
|
var topicsMap map[string]struct{}
|
||||||
if len(topics) > 0 {
|
if len(topics) > 0 {
|
||||||
topicsMap = make(map[string]struct{})
|
topicsMap = make(map[string]struct{})
|
||||||
@ -383,11 +438,12 @@ func newTracerWrapper(tr pubsub.EventTracer, topics ...string) pubsub.EventTrace
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return &tracerWrapper{tr: tr, topics: topicsMap}
|
return &tracerWrapper{lp2pTracer: lp2pTracer, lotusTracer: lotusTracer, topics: topicsMap}
|
||||||
}
|
}
|
||||||
|
|
||||||
type tracerWrapper struct {
|
type tracerWrapper struct {
|
||||||
tr pubsub.EventTracer
|
lp2pTracer pubsub.EventTracer
|
||||||
|
lotusTracer pubsub.EventTracer
|
||||||
topics map[string]struct{}
|
topics map[string]struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -406,33 +462,61 @@ func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
|
|||||||
switch evt.GetType() {
|
switch evt.GetType() {
|
||||||
case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
|
case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
|
||||||
stats.Record(context.TODO(), metrics.PubsubPublishMessage.M(1))
|
stats.Record(context.TODO(), metrics.PubsubPublishMessage.M(1))
|
||||||
if trw.tr != nil && trw.traceMessage(evt.GetPublishMessage().GetTopic()) {
|
if trw.traceMessage(evt.GetPublishMessage().GetTopic()) {
|
||||||
trw.tr.Trace(evt)
|
if trw.lp2pTracer != nil {
|
||||||
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
|
case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
|
||||||
stats.Record(context.TODO(), metrics.PubsubDeliverMessage.M(1))
|
stats.Record(context.TODO(), metrics.PubsubDeliverMessage.M(1))
|
||||||
if trw.tr != nil && trw.traceMessage(evt.GetDeliverMessage().GetTopic()) {
|
if trw.traceMessage(evt.GetDeliverMessage().GetTopic()) {
|
||||||
trw.tr.Trace(evt)
|
if trw.lp2pTracer != nil {
|
||||||
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_REJECT_MESSAGE:
|
case pubsub_pb.TraceEvent_REJECT_MESSAGE:
|
||||||
stats.Record(context.TODO(), metrics.PubsubRejectMessage.M(1))
|
stats.Record(context.TODO(), metrics.PubsubRejectMessage.M(1))
|
||||||
case pubsub_pb.TraceEvent_DUPLICATE_MESSAGE:
|
case pubsub_pb.TraceEvent_DUPLICATE_MESSAGE:
|
||||||
stats.Record(context.TODO(), metrics.PubsubDuplicateMessage.M(1))
|
stats.Record(context.TODO(), metrics.PubsubDuplicateMessage.M(1))
|
||||||
case pubsub_pb.TraceEvent_JOIN:
|
case pubsub_pb.TraceEvent_JOIN:
|
||||||
if trw.tr != nil {
|
if trw.lp2pTracer != nil {
|
||||||
trw.tr.Trace(evt)
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_LEAVE:
|
case pubsub_pb.TraceEvent_LEAVE:
|
||||||
if trw.tr != nil {
|
if trw.lp2pTracer != nil {
|
||||||
trw.tr.Trace(evt)
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_GRAFT:
|
case pubsub_pb.TraceEvent_GRAFT:
|
||||||
if trw.tr != nil {
|
if trw.lp2pTracer != nil {
|
||||||
trw.tr.Trace(evt)
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_PRUNE:
|
case pubsub_pb.TraceEvent_PRUNE:
|
||||||
if trw.tr != nil {
|
if trw.lp2pTracer != nil {
|
||||||
trw.tr.Trace(evt)
|
trw.lp2pTracer.Trace(evt)
|
||||||
|
}
|
||||||
|
|
||||||
|
if trw.lotusTracer != nil {
|
||||||
|
trw.lotusTracer.Trace(evt)
|
||||||
}
|
}
|
||||||
case pubsub_pb.TraceEvent_RECV_RPC:
|
case pubsub_pb.TraceEvent_RECV_RPC:
|
||||||
stats.Record(context.TODO(), metrics.PubsubRecvRPC.M(1))
|
stats.Record(context.TODO(), metrics.PubsubRecvRPC.M(1))
|
||||||
|
92
node/modules/tracer/elasticsearch_transport.go
Normal file
92
node/modules/tracer/elasticsearch_transport.go
Normal file
@ -0,0 +1,92 @@
|
|||||||
|
package tracer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/url"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"github.com/elastic/go-elasticsearch/v7"
|
||||||
|
"github.com/elastic/go-elasticsearch/v7/esapi"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
ElasticSearch_INDEX_DEFAULT = "lotus-pubsub"
|
||||||
|
)
|
||||||
|
|
||||||
|
func NewElasticSearchTransport(connectionString string, elasticsearchIndex string) (TracerTransport, error) {
|
||||||
|
conUrl, err := url.Parse(connectionString)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
username := conUrl.User.Username()
|
||||||
|
password, _ := conUrl.User.Password()
|
||||||
|
cfg := elasticsearch.Config{
|
||||||
|
Addresses: []string{
|
||||||
|
conUrl.Scheme + "://" + conUrl.Host,
|
||||||
|
},
|
||||||
|
Username: username,
|
||||||
|
Password: password,
|
||||||
|
}
|
||||||
|
|
||||||
|
es, err := elasticsearch.NewClient(cfg)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
var esIndex string
|
||||||
|
if elasticsearchIndex != "" {
|
||||||
|
esIndex = elasticsearchIndex
|
||||||
|
} else {
|
||||||
|
esIndex = ElasticSearch_INDEX_DEFAULT
|
||||||
|
}
|
||||||
|
|
||||||
|
return &elasticSearchTransport{
|
||||||
|
cl: es,
|
||||||
|
esIndex: esIndex,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type elasticSearchTransport struct {
|
||||||
|
cl *elasticsearch.Client
|
||||||
|
esIndex string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (est *elasticSearchTransport) Transport(evt TracerTransportEvent) error {
|
||||||
|
var e interface{}
|
||||||
|
|
||||||
|
if evt.lotusTraceEvent != nil {
|
||||||
|
e = *evt.lotusTraceEvent
|
||||||
|
} else if evt.pubsubTraceEvent != nil {
|
||||||
|
e = *evt.pubsubTraceEvent
|
||||||
|
} else {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
jsonEvt, err := json.Marshal(e)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("error while marshaling event: %s", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
req := esapi.IndexRequest{
|
||||||
|
Index: est.esIndex,
|
||||||
|
Body: strings.NewReader(string(jsonEvt)),
|
||||||
|
Refresh: "true",
|
||||||
|
}
|
||||||
|
|
||||||
|
// Perform the request with the client.
|
||||||
|
res, err := req.Do(context.Background(), est.cl)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer res.Body.Close()
|
||||||
|
|
||||||
|
if res.IsError() {
|
||||||
|
return fmt.Errorf("[%s] Error indexing document ID=%s", res.Status(), req.DocumentID)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
41
node/modules/tracer/json_transport.go
Normal file
41
node/modules/tracer/json_transport.go
Normal file
@ -0,0 +1,41 @@
|
|||||||
|
package tracer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
)
|
||||||
|
|
||||||
|
type jsonTracerTransport struct {
|
||||||
|
out *os.File
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewJsonTracerTransport(file string) (TracerTransport, error) {
|
||||||
|
out, err := os.OpenFile(file, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0660)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return &jsonTracerTransport{
|
||||||
|
out: out,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (jtt *jsonTracerTransport) Transport(evt TracerTransportEvent) error {
|
||||||
|
var e interface{}
|
||||||
|
if evt.lotusTraceEvent != nil {
|
||||||
|
e = *evt.lotusTraceEvent
|
||||||
|
} else if evt.pubsubTraceEvent != nil {
|
||||||
|
e = *evt.pubsubTraceEvent
|
||||||
|
} else {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
jsonEvt, err := json.Marshal(e)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("error while marshaling event: %s", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = jtt.out.WriteString(string(jsonEvt) + "\n")
|
||||||
|
return err
|
||||||
|
}
|
90
node/modules/tracer/tracer.go
Normal file
90
node/modules/tracer/tracer.go
Normal file
@ -0,0 +1,90 @@
|
|||||||
|
package tracer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
logging "github.com/ipfs/go-log/v2"
|
||||||
|
"github.com/libp2p/go-libp2p-core/peer"
|
||||||
|
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
||||||
|
pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
|
||||||
|
)
|
||||||
|
|
||||||
|
var log = logging.Logger("lotus-tracer")
|
||||||
|
|
||||||
|
func NewLotusTracer(tt []TracerTransport, pid peer.ID, sourceAuth string) LotusTracer {
|
||||||
|
return &lotusTracer{
|
||||||
|
tt: tt,
|
||||||
|
pid: pid,
|
||||||
|
sa: sourceAuth,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type lotusTracer struct {
|
||||||
|
tt []TracerTransport
|
||||||
|
pid peer.ID
|
||||||
|
sa string
|
||||||
|
}
|
||||||
|
|
||||||
|
const (
|
||||||
|
TraceEvent_PEER_SCORES pubsub_pb.TraceEvent_Type = 100
|
||||||
|
)
|
||||||
|
|
||||||
|
type LotusTraceEvent struct {
|
||||||
|
Type pubsub_pb.TraceEvent_Type `json:"type,omitempty"`
|
||||||
|
PeerID string `json:"peerID,omitempty"`
|
||||||
|
Timestamp *int64 `json:"timestamp,omitempty"`
|
||||||
|
PeerScore TraceEvent_PeerScore `json:"peerScore,omitempty"`
|
||||||
|
SourceAuth string `json:"sourceAuth,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type TraceEvent_PeerScore struct {
|
||||||
|
PeerID string `json:"peerID"`
|
||||||
|
Score float32 `json:"score"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type LotusTracer interface {
|
||||||
|
Trace(evt *pubsub_pb.TraceEvent)
|
||||||
|
TraceLotusEvent(evt *LotusTraceEvent)
|
||||||
|
|
||||||
|
PeerScores(scores map[peer.ID]*pubsub.PeerScoreSnapshot)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (lt *lotusTracer) PeerScores(scores map[peer.ID]*pubsub.PeerScoreSnapshot) {
|
||||||
|
now := time.Now().UnixNano()
|
||||||
|
for pid, score := range scores {
|
||||||
|
evt := &LotusTraceEvent{
|
||||||
|
Type: *TraceEvent_PEER_SCORES.Enum(),
|
||||||
|
PeerID: lt.pid.Pretty(),
|
||||||
|
Timestamp: &now,
|
||||||
|
SourceAuth: lt.sa,
|
||||||
|
PeerScore: TraceEvent_PeerScore{PeerID: pid.Pretty(), Score: float32(score.Score)},
|
||||||
|
}
|
||||||
|
|
||||||
|
lt.TraceLotusEvent(evt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (lt *lotusTracer) TraceLotusEvent(evt *LotusTraceEvent) {
|
||||||
|
for _, t := range lt.tt {
|
||||||
|
err := t.Transport(TracerTransportEvent{
|
||||||
|
lotusTraceEvent: evt,
|
||||||
|
pubsubTraceEvent: nil,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("error while transporting peer scores: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
func (lt *lotusTracer) Trace(evt *pubsub_pb.TraceEvent) {
|
||||||
|
for _, t := range lt.tt {
|
||||||
|
err := t.Transport(TracerTransportEvent{
|
||||||
|
lotusTraceEvent: nil,
|
||||||
|
pubsubTraceEvent: evt,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("error while transporting trace event: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
12
node/modules/tracer/transport.go
Normal file
12
node/modules/tracer/transport.go
Normal file
@ -0,0 +1,12 @@
|
|||||||
|
package tracer
|
||||||
|
|
||||||
|
import pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
|
||||||
|
|
||||||
|
type TracerTransport interface {
|
||||||
|
Transport(evt TracerTransportEvent) error
|
||||||
|
}
|
||||||
|
|
||||||
|
type TracerTransportEvent struct {
|
||||||
|
lotusTraceEvent *LotusTraceEvent
|
||||||
|
pubsubTraceEvent *pubsub_pb.TraceEvent
|
||||||
|
}
|
14
tools/kibana/README.md
Normal file
14
tools/kibana/README.md
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
## Lotus Kibana Dashboard
|
||||||
|
|
||||||
|
This folder contains configuration files to create Kibana dashboards to track peer scores and block propagation
|
||||||
|
throughout Filecoin network.
|
||||||
|
|
||||||
|
### Importing dashboard
|
||||||
|
|
||||||
|
The peer score and block propagation dashboard configuration is imported via Kibana import saved object [functionality](https://www.elastic.co/guide/en/kibana/current/managing-saved-objects.html#managing-saved-objects-export-objects).
|
||||||
|
|
||||||
|
The index patterns will be created automatically when importing dashboards.
|
||||||
|
|
||||||
|
#### Custom index
|
||||||
|
|
||||||
|
By default, the dashboards target `lotus-pubsub` index which is the default one when running node. The index can be customised via edit on dashboard visualizations.
|
2
tools/kibana/block-propagation-dashboard.ndjson
Normal file
2
tools/kibana/block-propagation-dashboard.ndjson
Normal file
File diff suppressed because one or more lines are too long
3
tools/kibana/peer-scores-dashboard.ndjson
Normal file
3
tools/kibana/peer-scores-dashboard.ndjson
Normal file
@ -0,0 +1,3 @@
|
|||||||
|
{"attributes":{"fieldAttrs":"{}","fields":"[]","runtimeFieldMap":"{\"peerScore.weightedScore\":{\"type\":\"double\",\"script\":{\"source\":\"if (doc['type'].value == 100) {\\n def score = doc['peerScore.score'].value;\\n if (doc['sourceAuth'] == \\\"<password>\\\") {\\n\\n emit(score * 1.2)\\n } else {\\n emit(score)\\n }\\n}\\n\\n\"}},\"sourceAuth\":{\"type\":\"keyword\"}}","title":"lotus-pubsub*","typeMeta":"{}"},"coreMigrationVersion":"7.14.1","id":"2c407db0-1acb-11ec-99f4-75d57f0cd0d8","migrationVersion":{"index-pattern":"7.11.0"},"references":[],"type":"index-pattern","updated_at":"2021-09-24T12:03:02.575Z","version":"WzcwMzksMV0="}
|
||||||
|
{"attributes":{"description":"Average peer score table per node peerID","hits":0,"kibanaSavedObjectMeta":{"searchSourceJSON":"{\"query\":{\"query\":\"\",\"language\":\"kuery\"},\"filter\":[]}"},"optionsJSON":"{\"useMargins\":true,\"syncColors\":false,\"hidePanelTitles\":false}","panelsJSON":"[{\"version\":\"7.14.1\",\"type\":\"lens\",\"gridData\":{\"x\":0,\"y\":0,\"w\":48,\"h\":43,\"i\":\"cddc98a5-45f3-4ba7-a7c6-6b9d216b19da\"},\"panelIndex\":\"cddc98a5-45f3-4ba7-a7c6-6b9d216b19da\",\"embeddableConfig\":{\"attributes\":{\"title\":\"\",\"type\":\"lens\",\"visualizationType\":\"lnsDatatable\",\"state\":{\"datasourceStates\":{\"indexpattern\":{\"layers\":{\"666e2f39-8868-45ad-b747-fe124830b0ae\":{\"columns\":{\"504c50bd-14c1-4119-820e-c961866fc3b4\":{\"label\":\"peerID\",\"dataType\":\"string\",\"operationType\":\"terms\",\"scale\":\"ordinal\",\"sourceField\":\"peerScore.peerID.keyword\",\"isBucketed\":true,\"params\":{\"size\":100,\"orderBy\":{\"type\":\"column\",\"columnId\":\"ec82c5f7-b3c4-4715-8646-a8fe584400fc\"},\"orderDirection\":\"desc\",\"otherBucket\":false,\"missingBucket\":false},\"customLabel\":true},\"ec82c5f7-b3c4-4715-8646-a8fe584400fc\":{\"label\":\"Score\",\"dataType\":\"number\",\"operationType\":\"average\",\"sourceField\":\"peerScore.score\",\"isBucketed\":false,\"scale\":\"ratio\",\"customLabel\":true},\"e22a19e8-1d71-43d6-9ca0-b30ddd423447\":{\"label\":\"Weighted Score\",\"dataType\":\"number\",\"operationType\":\"average\",\"sourceField\":\"peerScore.weightedScore\",\"isBucketed\":false,\"scale\":\"ratio\",\"customLabel\":true}},\"columnOrder\":[\"504c50bd-14c1-4119-820e-c961866fc3b4\",\"ec82c5f7-b3c4-4715-8646-a8fe584400fc\",\"e22a19e8-1d71-43d6-9ca0-b30ddd423447\"],\"incompleteColumns\":{}}}}},\"visualization\":{\"columns\":[{\"isTransposed\":false,\"columnId\":\"504c50bd-14c1-4119-820e-c961866fc3b4\"},{\"isTransposed\":false,\"columnId\":\"ec82c5f7-b3c4-4715-8646-a8fe584400fc\",\"colorMode\":\"cell\",\"palette\":{\"type\":\"palette\",\"name\":\"positive\",\"params\":{\"stops\":[{\"color\":\"#d6e9e4\",\"stop\":20},{\"color\":\"#aed3ca\",\"stop\":40},{\"color\":\"#85bdb1\",\"stop\":60},{\"color\":\"#5aa898\",\"stop\":80},{\"color\":\"#209280\",\"stop\":100}]}}},{\"columnId\":\"e22a19e8-1d71-43d6-9ca0-b30ddd423447\",\"isTransposed\":false,\"colorMode\":\"cell\",\"palette\":{\"type\":\"palette\",\"name\":\"positive\",\"params\":{\"stops\":[{\"color\":\"#d6e9e4\",\"stop\":20},{\"color\":\"#aed3ca\",\"stop\":40},{\"color\":\"#85bdb1\",\"stop\":60},{\"color\":\"#5aa898\",\"stop\":80},{\"color\":\"#209280\",\"stop\":100}]}}}],\"layerId\":\"666e2f39-8868-45ad-b747-fe124830b0ae\"},\"query\":{\"query\":\"\",\"language\":\"kuery\"},\"filters\":[]},\"references\":[{\"type\":\"index-pattern\",\"id\":\"9890c040-17b7-11ec-99f4-75d57f0cd0d8\",\"name\":\"indexpattern-datasource-current-indexpattern\"},{\"type\":\"index-pattern\",\"id\":\"2c407db0-1acb-11ec-99f4-75d57f0cd0d8\",\"name\":\"indexpattern-datasource-layer-666e2f39-8868-45ad-b747-fe124830b0ae\"}]},\"hidePanelTitles\":false,\"enhancements\":{}},\"title\":\"Peer Scores\"}]","timeRestore":false,"title":"Peer Scores","version":1},"coreMigrationVersion":"7.14.1","id":"e7e4fd70-1acb-11ec-99f4-75d57f0cd0d8","migrationVersion":{"dashboard":"7.14.0"},"references":[{"id":"2c407db0-1acb-11ec-99f4-75d57f0cd0d8","name":"cddc98a5-45f3-4ba7-a7c6-6b9d216b19da:indexpattern-datasource-current-indexpattern","type":"index-pattern"},{"id":"2c407db0-1acb-11ec-99f4-75d57f0cd0d8","name":"cddc98a5-45f3-4ba7-a7c6-6b9d216b19da:indexpattern-datasource-layer-666e2f39-8868-45ad-b747-fe124830b0ae","type":"index-pattern"}],"type":"dashboard","updated_at":"2021-09-24T11:59:46.187Z","version":"WzY5NjcsMV0="}
|
||||||
|
{"excludedObjects":[],"excludedObjectsCount":0,"exportedCount":2,"missingRefCount":0,"missingReferences":[]}
|
Loading…
Reference in New Issue
Block a user