peermanager plugin

plugins/peermanager/kafka.go

@@ -0,0 +1,46 @@
+package peermanager
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+	"github.com/openrelayxyz/cardinal-streams/transports"
+)
+
+func strPtr(x string) *string {
+	return &x
+}
+
+func createProducer(broker, topic string) (sarama.AsyncProducer, error) {
+
+	sessionBrokers, sessionKafkaConfig = transports.ParseKafkaURL(strings.TrimPrefix(broker, "kafka://"))
+	configEntries := make(map[string]*string)
+	configEntries["retention.ms"] = strPtr("3600000")
+
+	if err := transports.CreateTopicIfDoesNotExist(strings.TrimPrefix(broker, "kafka://"), topic, 1, configEntries); err != nil {
+		panic(fmt.Sprintf("Could not create topic %v on broker %v: %v", topic, broker, err.Error()))
+	}
+
+	producer, err := sarama.NewAsyncProducer(sessionBrokers, sessionKafkaConfig)
+	if err != nil {
+		panic(fmt.Sprintf("Could not setup producer, peer manager plugin: %v", err.Error()))
+	}
+
+	return producer, nil
+}
+
+func createConsumer(broker, topic string) (sarama.PartitionConsumer, error) {
+
+	consumer, err := sarama.NewConsumer(sessionBrokers, sessionKafkaConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest)
+	if err != nil {
+		return nil, err
+	}
+
+	return partitionConsumer, nil
+}

plugins/peermanager/main.go

@@ -0,0 +1,108 @@
+package peermanager
+
+import (
+	"fmt"
+
+	"github.com/Shopify/sarama"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/openrelayxyz/xplugeth"
+	"github.com/openrelayxyz/xplugeth/hooks/initialize"
+	"github.com/openrelayxyz/xplugeth/types"
+)
+
+type peerManagerConfig struct {
+	BrokerURL string `yaml:"peerManager.broker.url"`
+}
+
+var cfg *peerManagerConfig
+
+var (
+	httpApiFlagName    = "http.api"
+	sessionStack       node.Node
+	sessionBrokers     []string
+	sessionKafkaConfig *sarama.Config
+	nodes              = make(chan string, 5)
+	exit               = make(chan struct{}, 1)
+	peerBroker         string
+)
+
+type peerManagerModule struct{}
+
+func init() {
+	xplugeth.RegisterModule[peerManagerModule]("peerManagerModule")
+}
+
+func (*peerManagerModule) InitializeNode(s *node.Node, b types.Backend) {
+	sessionStack = *s
+	var ok bool
+	cfg, ok = xplugeth.GetConfig[peerManagerConfig]("peerManager")
+	peerBroker = cfg.BrokerURL
+	if !ok {
+		log.Warn("did not acqire config, example plugin, all values set to default")
+	}
+	log.Info("Initialized node, peer manager plugin")
+}
+
+func (*peerManagerModule) Blockchain() {
+	if peerBroker == "" {
+		panic(fmt.Sprintf("no broker provided for peer manager plugin"))
+	}
+	go peeringSequence()
+}
+
+func peeringSequence() {
+	sessionPeerService, err := getPeerManager()
+	if err != nil {
+		log.Error("session peer service unavailable, peer manager plugin", "err", err)
+		return
+	}
+
+	selfNode, err := sessionPeerService.getEnode()
+	if err != nil {
+		log.Error("error calling getEnode from sessionService, peer manager plugin", "err", err)
+	}
+
+	chainTopic, err := sessionPeerService.chainIdResolver()
+	if err != nil {
+		log.Error("Error aquiring chainID, peer manager plugin", "err", err)
+	}
+
+	producer, err := createProducer(peerBroker, chainTopic)
+	if err != nil {
+		log.Error("failed to acquire kafka producer, peer manager plugin", "err", err)
+		return
+	}
+
+	consumer, err := createConsumer(peerBroker, chainTopic)
+	if err != nil {
+		log.Error("failed to acquire kafka consumer, peer manager plugin", "err", err)
+		return
+	}
+
+	msg := &sarama.ProducerMessage{
+		Topic: chainTopic,
+		Value: sarama.StringEncoder(selfNode),
+	}
+
+	producer.Input() <- msg
+
+	go func() {
+		for message := range consumer.Messages() {
+			nodes <- string(message.Value)
+		}
+	}()
+
+	for message := range nodes {
+		if message == selfNode {
+			continue
+		} else {
+			sessionPeerService.attachPeers(message)
+		}
+	}
+}
+
+var (
+	_ initialize.Blockchain  = (*peerManagerModule)(nil)
+	_ initialize.Initializer = (*peerManagerModule)(nil)
+)

plugins/peermanager/peer-manager.go

@@ -0,0 +1,84 @@
+package peermanager
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type PeerManagerService struct {
+	client *rpc.Client
+}
+
+type pseudoNodeInfo struct {
+	Enode string
+}
+
+func getPeerManager() (*PeerManagerService, error) {
+	c := sessionStack.Attach()
+
+	s := &PeerManagerService{
+		client: c,
+	}
+	return s, nil
+}
+
+func (service *PeerManagerService) getEnode() (string, error) {
+	var enode pseudoNodeInfo
+	err := service.client.Call(&enode, "admin_nodeInfo")
+	if err != nil {
+		return "", err
+	}
+	return enode.Enode, nil
+}
+
+func (service *PeerManagerService) attachPeers(peer string) {
+
+	var addTrustedPeerResult bool
+	err := service.client.Call(&addTrustedPeerResult, "admin_addTrustedPeer", peer)
+	if err != nil {
+		log.Error("error calling admin_addTrustedPeer, peer manager plugin", "peer", peer, "err", err)
+	}
+	if !addTrustedPeerResult {
+		log.Error("addTrustedPeer returned false, peer manager plugin", "peer", peer, "err", err)
+	}
+
+	var addPeerResult bool
+	err = service.client.Call(&addPeerResult, "admin_addPeer", peer)
+	if err != nil {
+		log.Error("error calling admin_addPeer, peer manager plugin", "peer", peer, "err", err)
+	}
+	if !addPeerResult {
+		log.Error("addPeer returned false, peer manager plugin", "peer", peer, "err", err)
+	}
+	log.Info("added peer, peer manager plugin", "added", peer)
+}
+
+func (service *PeerManagerService) chainIdResolver() (string, error) {
+
+	var chainID string
+	err := service.client.Call(&chainID, "eth_chainId")
+	if err != nil {
+		return "", err
+	}
+	var result string
+	switch chainID {
+	case "0x1":
+		result = "mainnet"
+	case "0x3d":
+		result = "etc"
+	case "0x4268":
+		result = "holesky"
+	case "0xaa36a7":
+		result = "sepolia"
+	case "0x89":
+		result = "polygon"
+	case "0x13881":
+		result = "mumbai"
+	default:
+		log.Warn("unknown chain, chainID could not be resolved, peer manager plugin")
+		result = chainID
+	}
+	return fmt.Sprintf("peers-%v", result), nil
+}