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

Merge pull request #7398 from ChainSafe/libp2p-pubsub-tracer

Browse Source 
Lotus extended pubsub tracer
This commit is contained in:
Łukasz Magiera
2022-12-08 12:14:56 +01:00
committed by GitHub
parent 5a3b6becf4 62e98c1964
commit a10c014ba6
16 changed files with 684 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/config/doc_gen.go
+29
View File
@@ -788,6 +788,35 @@ Type: Array of multiaddress peerinfo strings, must include peerid (/p2p/12D3K...
Comment: ``,
},
{
Name: "JsonTracer",
Type: "string",
Comment: `Path to file that will be used to output tracer content in JSON format.
If present tracer will save data to defined file.
Format: file path`,
},
{
Name: "ElasticSearchTracer",
Type: "string",
Comment: `Connection string for elasticsearch instance.
If present tracer will save data to elasticsearch.
Format: https://<username>:<password>@<elasticsearch_url>:<port>/`,
},
{
Name: "ElasticSearchIndex",
Type: "string",
Comment: `Name of elasticsearch index that will be used to save tracer data.
This property is used only if ElasticSearchTracer propery is set.`,
},
{
Name: "TracerSourceAuth",
Type: "string",
Comment: `Auth token that will be passed with logs to elasticsearch - used for weighted peers score.`,
},
},
"RetrievalPricing": []DocField{
{
node/config/types.go
+13
View File
@@ -563,6 +563,19 @@ type Pubsub struct {
DirectPeers []string
IPColocationWhitelist []string
RemoteTracer string
// Path to file that will be used to output tracer content in JSON format.
// If present tracer will save data to defined file.
// Format: file path
JsonTracer string
// Connection string for elasticsearch instance.
// If present tracer will save data to elasticsearch.
// Format: https://<username>:<password>@<elasticsearch_url>:<port>/
ElasticSearchTracer string
// Name of elasticsearch index that will be used to save tracer data.
// This property is used only if ElasticSearchTracer propery is set.
ElasticSearchIndex string
// Auth token that will be passed with logs to elasticsearch - used for weighted peers score.
TracerSourceAuth string
}
type Chainstore struct {
node/modules/lp2p/pubsub.go
+112 -19
View File
@@ -21,6 +21,7 @@ import (
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/modules/tracer"
)
func init() {
@@ -49,6 +50,30 @@ func ScoreKeeper() *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 {
fx.In
Mctx helpers.MetricsCtx
@@ -291,7 +316,6 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
OpportunisticGraftThreshold: OpportunisticGraftScoreThreshold,
},
),
pubsub.WithPeerScoreInspect(in.Sk.Update, 10*time.Second),
}
// enable Peer eXchange on bootstrappers
@@ -361,6 +385,27 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
pubsub.NewAllowlistSubscriptionFilter(allowTopics...),
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
if in.Cfg.RemoteTracer != "" {
a, err := ma.NewMultiaddr(in.Cfg.RemoteTracer)
@@ -378,12 +423,18 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
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.WithPeerScoreInspect(pst.UpdatePeerScore, 10*time.Second))
} else {
// still instantiate a tracer for collecting metrics
trw := newTracerWrapper(nil)
trw := newTracerWrapper(nil, lt)
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...)
@@ -394,7 +445,11 @@ func HashMsgId(m *pubsub_pb.Message) string {
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{}
if len(topics) > 0 {
topicsMap = make(map[string]struct{})
@@ -403,12 +458,13 @@ 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 {
tr pubsub.EventTracer
topics map[string]struct{}
lp2pTracer pubsub.EventTracer
lotusTracer pubsub.EventTracer
topics map[string]struct{}
}
func (trw *tracerWrapper) traceMessage(topic string) bool {
@@ -426,33 +482,70 @@ func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
switch evt.GetType() {
case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
stats.Record(context.TODO(), metrics.PubsubPublishMessage.M(1))
if trw.tr != nil && trw.traceMessage(evt.GetPublishMessage().GetTopic()) {
trw.tr.Trace(evt)
if trw.traceMessage(evt.GetPublishMessage().GetTopic()) {
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
}
case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
stats.Record(context.TODO(), metrics.PubsubDeliverMessage.M(1))
if trw.tr != nil && trw.traceMessage(evt.GetDeliverMessage().GetTopic()) {
trw.tr.Trace(evt)
if trw.traceMessage(evt.GetDeliverMessage().GetTopic()) {
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
}
case pubsub_pb.TraceEvent_REJECT_MESSAGE:
stats.Record(context.TODO(), metrics.PubsubRejectMessage.M(1))
if trw.traceMessage(evt.GetRejectMessage().GetTopic()) {
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
}
case pubsub_pb.TraceEvent_DUPLICATE_MESSAGE:
stats.Record(context.TODO(), metrics.PubsubDuplicateMessage.M(1))
case pubsub_pb.TraceEvent_JOIN:
if trw.tr != nil {
trw.tr.Trace(evt)
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
case pubsub_pb.TraceEvent_LEAVE:
if trw.tr != nil {
trw.tr.Trace(evt)
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
case pubsub_pb.TraceEvent_GRAFT:
if trw.tr != nil {
trw.tr.Trace(evt)
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
case pubsub_pb.TraceEvent_PRUNE:
if trw.tr != nil {
trw.tr.Trace(evt)
if trw.lp2pTracer != nil {
trw.lp2pTracer.Trace(evt)
}
if trw.lotusTracer != nil {
trw.lotusTracer.Trace(evt)
}
case pubsub_pb.TraceEvent_RECV_RPC:
stats.Record(context.TODO(), metrics.PubsubRecvRPC.M(1))
node/modules/tracer/elasticsearch_transport.go
+97
View File
@@ -0,0 +1,97 @@
package tracer
import (
"context"
"encoding/json"
"fmt"
"net/url"
"strings"
"github.com/elastic/go-elasticsearch/v7"
"github.com/elastic/go-elasticsearch/v7/esapi"
)
const (
ElasticSearchDefaultIndex = "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 = ElasticSearchDefaultIndex
}
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
}
err = res.Body.Close()
if err != nil {
return err
}
if res.IsError() {
return fmt.Errorf("[%s] Error indexing document ID=%s", res.Status(), req.DocumentID)
}
return nil
}
node/modules/tracer/json_transport.go
+41
View 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
}
node/modules/tracer/tracer.go
+120
View File
@@ -0,0 +1,120 @@
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 (
TraceEventPeerScores 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 TraceEventPeerScore `json:"peerScore,omitempty"`
SourceAuth string `json:"sourceAuth,omitempty"`
}
type TopicScore struct {
Topic string `json:"topic"`
TimeInMesh time.Duration `json:"timeInMesh"`
FirstMessageDeliveries float64 `json:"firstMessageDeliveries"`
MeshMessageDeliveries float64 `json:"meshMessageDeliveries"`
InvalidMessageDeliveries float64 `json:"invalidMessageDeliveries"`
}
type TraceEventPeerScore struct {
PeerID string `json:"peerID"`
Score float64 `json:"score"`
AppSpecificScore float64 `json:"appSpecificScore"`
IPColocationFactor float64 `json:"ipColocationFactor"`
BehaviourPenalty float64 `json:"behaviourPenalty"`
Topics []TopicScore `json:"topics"`
}
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 {
var topics []TopicScore
for topic, snapshot := range score.Topics {
topics = append(topics, TopicScore{
Topic: topic,
TimeInMesh: snapshot.TimeInMesh,
FirstMessageDeliveries: snapshot.FirstMessageDeliveries,
MeshMessageDeliveries: snapshot.MeshMessageDeliveries,
InvalidMessageDeliveries: snapshot.InvalidMessageDeliveries,
})
}
evt := &LotusTraceEvent{
Type: *TraceEventPeerScores.Enum(),
PeerID: lt.pid.Pretty(),
Timestamp: &now,
SourceAuth: lt.sa,
PeerScore: TraceEventPeerScore{
PeerID: pid.Pretty(),
Score: score.Score,
AppSpecificScore: score.AppSpecificScore,
IPColocationFactor: score.IPColocationFactor,
BehaviourPenalty: score.BehaviourPenalty,
Topics: topics,
},
}
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)
}
}
}
node/modules/tracer/tracer_test.go
+109
View File
@@ -0,0 +1,109 @@
package tracer
import (
"testing"
"time"
"github.com/libp2p/go-libp2p-core/peer"
pubsub "github.com/libp2p/go-libp2p-pubsub"
pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
"github.com/stretchr/testify/require"
)
type testTracerTransport struct {
t *testing.T
executeTest func(t *testing.T, evt TracerTransportEvent)
}
const peerIDA peer.ID = "12D3KooWAbSVMgRejb6ECg6fRTkCPGCfu8396msZVryu8ivcz44G"
func NewTestTraceTransport(t *testing.T, executeTest func(t *testing.T, evt TracerTransportEvent)) TracerTransport {
return &testTracerTransport{
t: t,
executeTest: executeTest,
}
}
func (ttt *testTracerTransport) Transport(evt TracerTransportEvent) error {
ttt.executeTest(ttt.t, evt)
return nil
}
func TestTracer_PeerScores(t *testing.T) {
testTransport := NewTestTraceTransport(t, func(t *testing.T, evt TracerTransportEvent) {
require.Equal(t, peerIDA.Pretty(), evt.lotusTraceEvent.PeerID)
require.Equal(t, "source-auth-token-test", evt.lotusTraceEvent.SourceAuth)
require.Equal(t, float64(32), evt.lotusTraceEvent.PeerScore.Score)
n := time.Now().UnixNano()
require.LessOrEqual(t, *evt.lotusTraceEvent.Timestamp, n)
require.Equal(t, peerIDA.Pretty(), evt.lotusTraceEvent.PeerScore.PeerID)
require.Equal(t, 1, len(evt.lotusTraceEvent.PeerScore.Topics))
topic := evt.lotusTraceEvent.PeerScore.Topics[0]
require.Equal(t, "topicA", topic.Topic)
require.Equal(t, float64(100), topic.FirstMessageDeliveries)
})
lt := NewLotusTracer(
[]TracerTransport{testTransport},
peerIDA,
"source-auth-token-test",
)
topics := make(map[string]*pubsub.TopicScoreSnapshot)
topics["topicA"] = &pubsub.TopicScoreSnapshot{
FirstMessageDeliveries: float64(100),
}
m := make(map[peer.ID]*pubsub.PeerScoreSnapshot)
m[peerIDA] = &pubsub.PeerScoreSnapshot{
Score: float64(32),
Topics: topics,
}
lt.PeerScores(m)
}
func TestTracer_PubSubTrace(t *testing.T) {
n := time.Now().Unix()
testTransport := NewTestTraceTransport(t, func(t *testing.T, evt TracerTransportEvent) {
require.Equal(t, []byte(peerIDA), evt.pubsubTraceEvent.PeerID)
require.Equal(t, &n, evt.pubsubTraceEvent.Timestamp)
})
lt := NewLotusTracer(
[]TracerTransport{testTransport},
"pid",
"source-auth",
)
lt.Trace(&pubsub_pb.TraceEvent{
PeerID: []byte(peerIDA),
Timestamp: &n,
})
}
func TestTracer_MultipleTransports(t *testing.T) {
testTransportA := NewTestTraceTransport(t, func(t *testing.T, evt TracerTransportEvent) {
require.Equal(t, []byte(peerIDA), evt.pubsubTraceEvent.PeerID)
})
testTransportB := NewTestTraceTransport(t, func(t *testing.T, evt TracerTransportEvent) {
require.Equal(t, []byte(peerIDA), evt.pubsubTraceEvent.PeerID)
})
executeTest := NewLotusTracer(
[]TracerTransport{testTransportA, testTransportB},
"pid",
"source-auth",
)
executeTest.Trace(&pubsub_pb.TraceEvent{
PeerID: []byte(peerIDA),
})
}
node/modules/tracer/transport.go
+12
View 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
}