Merge pull request #135 from gfanton/feat/add-eventbus-metrics

Author: gfanton

accesscontroller/orbitdb/accesscontroller_orbitdb.go

@@ -16,6 +16,7 @@ import (
 
 	cid "github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p/core/event"
+	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 	"go.uber.org/zap"
 )
 
@@ -217,7 +218,7 @@ func (o *orbitDBAccessController) Load(ctx context.Context, address string) erro
 		new(stores.EventWrite),
 		new(stores.EventReady),
 		new(stores.EventReplicated),
-	})
+	}, eventbus.Name("odb/load"))
 	if err != nil {
 		return fmt.Errorf("unable subscribe to store events: %w", err)
 	}

baseorbitdb/orbitdb.go

@@ -813,7 +813,8 @@ func (o *orbitDB) EventBus() event.Bus {
 }
 
 func (o *orbitDB) monitorDirectChannel(ctx context.Context, bus event.Bus) error {
-	sub, err := bus.Subscribe(new(iface.EventPubSubPayload), eventbus.BufSize(128))
+	sub, err := bus.Subscribe(new(iface.EventPubSubPayload),
+		eventbus.BufSize(128), eventbus.Name("odb/monitor-direct-channel"))
 	if err != nil {
 		return fmt.Errorf("unable to init pubsub subscriber: %w", err)
 	}

stores/basestore/base_store.go

@@ -268,7 +268,8 @@ func (b *BaseStore) InitBaseStore(ipfs coreapi.CoreAPI, identity *identityprovid
 
 	b.options = options
 
-	sub, err := b.replicator.EventBus().Subscribe(replicator.Events, eventbus.BufSize(128))
+	sub, err := b.replicator.EventBus().Subscribe(replicator.Events,
+		eventbus.BufSize(128), eventbus.Name("odb/base-store-main-loop"))
 	if err != nil {
 		return fmt.Errorf("unable to subscribe to replicator events: %w", err)
 	}
@@ -1008,7 +1009,7 @@ func (b *BaseStore) replicate() error {
 }
 
 func (b *BaseStore) storeListener(topic iface.PubSubTopic) error {
-	sub, err := b.EventBus().Subscribe(new(stores.EventWrite))
+	sub, err := b.EventBus().Subscribe(new(stores.EventWrite), eventbus.Name("odb/store-listener"))
 	if err != nil {
 		return fmt.Errorf("unable to init event bus: %w", err)
 	}