Merge pull request #7 from ChainSafe/mmuftic/elastic-search-transport

Add Elasticsearch transport

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/drand/drand v1.2.1
 	github.com/drand/kyber v1.1.4
 	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/gosigar v0.12.0
 	github.com/etclabscore/go-openrpc-reflect v0.0.36

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/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
 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/go.mod h1:i1ZYdU10oLNfRzq4vq62BEwD2fH8KaWh6eh0ikPT9F0=
 github.com/elastic/go-windows v1.0.0 h1:qLURgZFkkrYyTTkvYpsZIgf83AUsdIHfvlJaqaZ7aSY=

node/config/doc_gen.go

@@ -531,6 +531,12 @@ Type: Array of multiaddress peerinfo strings, must include peerid (/p2p/12D3K...
 			Name: "JsonTracerFile",
 			Type: "string",
 
+			Comment: ``,
+		},
+		{
+			Name: "ElasticSearchTracer",
+			Type: "string",
+
 			Comment: ``,
 		},
 	},

node/config/types.go

@@ -310,6 +310,7 @@ type Pubsub struct {
 	IPColocationWhitelist []string
 	RemoteTracer          string
 	JsonTracerFile        string
+	ElasticSearchTracer   string
 }
 
 type Chainstore struct {

node/modules/lp2p/pubsub.go

@@ -366,16 +366,25 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 				pubsub.NewAllowlistSubscriptionFilter(allowTopics...),
 				100)))
 
-	var lt tracer.LotusTracer
+	var transports []tracer.TracerTransport
 	if in.Cfg.JsonTracerFile != "" {
 		out, err := os.OpenFile(in.Cfg.JsonTracerFile, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0660)
 		if err != nil {
 			return nil, err
 		}
-
 		jsonTransport := tracer.NewJsonTracerTransport(out)
-		lt = tracer.NewLotusTracer(jsonTransport, in.Host.ID())
+		transports = append(transports, jsonTransport)
 	}
+	if in.Cfg.ElasticSearchTracer != "" {
+		elasticSearchTransport, err := tracer.NewElasticSearchTransport(
+			in.Cfg.ElasticSearchTracer,
+		)
+		if err != nil {
+			return nil, err
+		}
+		transports = append(transports, elasticSearchTransport)
+	}
+	lt := tracer.NewLotusTracer(transports, in.Host.ID())
 
 	// tracer
 	if in.Cfg.RemoteTracer != "" {

node/modules/tracer/elasticsearch_transport.go

@@ -0,0 +1,85 @@
+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 = "pubsub"
+
+	ElasticSearch_DOC_LOTUS  = "doc_lotus"
+	ElasticSearch_DOC_PUBSUB = "doc_pubsub"
+)
+
+func NewElasticSearchTransport(connectionString string) (TracerTransport, error) {
+	conUrl, err := url.Parse(connectionString)
+
+	username := conUrl.User.Username()
+	password, _ := conUrl.User.Password()
+	cfg := elasticsearch.Config{
+		Addresses: []string{
+			"https://" + conUrl.Host,
+		},
+		Username: username,
+		Password: password,
+	}
+
+	es, err := elasticsearch.NewClient(cfg)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &elasticSearchTransport{
+		cl: es,
+	}, nil
+}
+
+type elasticSearchTransport struct {
+	cl *elasticsearch.Client
+}
+
+func (est *elasticSearchTransport) Transport(evt TracerTransportEvent) error {
+	var e interface{}
+	var docId string
+	if evt.lotusTraceEvent != nil {
+		e = *evt.lotusTraceEvent
+		docId = ElasticSearch_DOC_LOTUS
+	} else if evt.pubsubTraceEvent != nil {
+		e = *evt.pubsubTraceEvent
+		docId = ElasticSearch_DOC_PUBSUB
+	} else {
+		return nil
+	}
+
+	jsonEvt, err := json.Marshal(e)
+	if err != nil {
+		return fmt.Errorf("error while marshaling event: %s", err)
+	}
+
+	req := esapi.IndexRequest{
+		Index:      ElasticSearch_INDEX,
+		DocumentID: docId,
+		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
+}

node/modules/tracer/json_transport.go

@@ -1,6 +1,8 @@
 package tracer
 
 import (
+	"encoding/json"
+	"fmt"
 	"os"
 )
 
@@ -14,7 +16,21 @@ func NewJsonTracerTransport(out *os.File) TracerTransport {
 	}
 }
 
-func (jtt *jsonTracerTransport) Transport(jsonEvent []byte) error {
-	_, err := jtt.out.Write(jsonEvent)
+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.Write(jsonEvt)
 	return err
 }

node/modules/tracer/tracer.go

@@ -1,7 +1,6 @@
 package tracer
 
 import (
-	"encoding/json"
 	"time"
 
 	logging "github.com/ipfs/go-log/v2"
@@ -12,7 +11,7 @@ import (
 
 var log = logging.Logger("lotus-tracer")
 
-func NewLotusTracer(tt TracerTransport, pid peer.ID) LotusTracer {
+func NewLotusTracer(tt []TracerTransport, pid peer.ID) LotusTracer {
 	return &lotusTracer{
 		tt:  tt,
 		pid: pid,
@@ -20,7 +19,7 @@ func NewLotusTracer(tt TracerTransport, pid peer.ID) LotusTracer {
 }
 
 type lotusTracer struct {
-	tt  TracerTransport
+	tt  []TracerTransport
 	pid peer.ID
 }
 
@@ -61,27 +60,26 @@ func (lt *lotusTracer) PeerScores(scores map[peer.ID]*pubsub.PeerScoreSnapshot)
 }
 
 func (lt *lotusTracer) TraceLotusEvent(evt *LotusTraceEvent) {
-	jsonEvent, err := json.Marshal(evt)
-	if err != nil {
-		log.Errorf("error while marshaling peer score: %s", err)
-		return
+	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)
+		}
 	}
 
-	err = lt.tt.Transport(jsonEvent)
-	if err != nil {
-		log.Errorf("error while transporting peer scores: %s", err)
-	}
 }
 
 func (lt *lotusTracer) Trace(evt *pubsub_pb.TraceEvent) {
-	jsonEvent, err := json.Marshal(evt)
-	if err != nil {
-		log.Errorf("error while marshaling tracer event: %s", err)
-		return
-	}
-
-	err = lt.tt.Transport(jsonEvent)
-	if err != nil {
-		log.Errorf("error while transporting trace event: %s", err)
+	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/transport.go

@@ -1,5 +1,12 @@
 package tracer
 
+import pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
+
 type TracerTransport interface {
-	Transport(jsonEvent []byte) error
+	Transport(evt TracerTransportEvent) error
+}
+
+type TracerTransportEvent struct {
+	lotusTraceEvent  *LotusTraceEvent
+	pubsubTraceEvent *pubsub_pb.TraceEvent
 }