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

@@ -375,6 +375,9 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 
 		jsonTransport := tracer.NewJsonTracerTransport(out)
 		lt = tracer.NewLotusTracer(jsonTransport, in.Host.ID())
+	} else if in.Cfg.ElasticSearchTracer != "" {
+		elasticSearchTransport := tracer.NewElasticSearchTransport()
+		lt = tracer.NewLotusTracer(elasticSearchTransport, in.Host.ID())
 	}
 
 	// tracer

node/modules/tracer/elasticsearch_transport.go

@@ -0,0 +1,59 @@
+package tracer
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/elastic/go-elasticsearch/v7"
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+	logging "github.com/ipfs/go-log/v2"
+)
+
+var rpclog = logging.Logger("elasticsearch")
+
+func NewElasticSearchTransport() TracerTransport {
+	es, err := elasticsearch.NewDefaultClient()
+
+	if err != nil {
+		rpclog.Fatalf("Error on creating elastic search client: %+v", err)
+	}
+
+	return &elasticSearchTransport{
+		cl: es,
+	}
+}
+
+type elasticSearchTransport struct {
+	cl *elasticsearch.Client
+}
+
+func (est *elasticSearchTransport) Transport(jsonEvent []byte) error {
+	req := esapi.IndexRequest{
+		Index:      "PeerScore",
+		DocumentID: "1", // todo
+		Body:       strings.NewReader(string(jsonEvent)),
+		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)
+	} else {
+		// Deserialize the response into a map.
+		var r map[string]interface{}
+		if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
+			return err
+		} else {
+			rpclog.Infof("[%s] %s; version=%d", res.Status(), r["result"], int(r["_version"].(float64)))
+		}
+	}
+	return nil
+}

node/modules/tracer/tracer.go

@@ -12,7 +12,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 +20,7 @@ func NewLotusTracer(tt TracerTransport, pid peer.ID) LotusTracer {
 }
 
 type lotusTracer struct {
-	tt  TracerTransport
+	tt  []TracerTransport
 	pid peer.ID
 }
 
@@ -67,10 +67,13 @@ func (lt *lotusTracer) TraceLotusEvent(evt *LotusTraceEvent) {
 		return
 	}
 
-	err = lt.tt.Transport(jsonEvent)
-	if err != nil {
-		log.Errorf("error while transporting peer scores: %s", err)
+	for _, t := range lt.tt {
+		err = t.Transport(jsonEvent)
+		if err != nil {
+			log.Errorf("error while transporting peer scores: %s", err)
+		}
 	}
+
 }
 
 func (lt *lotusTracer) Trace(evt *pubsub_pb.TraceEvent) {
@@ -80,8 +83,10 @@ func (lt *lotusTracer) Trace(evt *pubsub_pb.TraceEvent) {
 		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(jsonEvent)
+		if err != nil {
+			log.Errorf("error while transporting trace event: %s", err)
+		}
 	}
 }