relayer metrics

nil/services/relayer/internal/l1/event_listener.go

@@ -51,6 +51,7 @@ type EventListener struct {
 	clock           clockwork.Clock
 
 	config       *EventListenerConfig
+	metrics      EventListenerMetrics
 	eventStorage *EventStorage
 
 	state struct {
@@ -70,6 +71,7 @@ func NewEventListener(
 	ethClient EthClient,
 	contractClient L1Contract,
 	storage *EventStorage,
+	metrics EventListenerMetrics,
 	logger logging.Logger,
 ) (*EventListener, error) {
 	el := &EventListener{
@@ -78,6 +80,7 @@ func NewEventListener(
 		clock:           clock,
 		config:          config,
 		eventStorage:    storage,
+		metrics:         metrics,
 	}
 
 	el.state.emitter = make(chan struct{}, config.EmitEventCapacity)
@@ -161,8 +164,6 @@ func (el *EventListener) subscriber(ctx context.Context, eventCh chan<- *L1Messa
 			Err(err).
 			Msg("failed to subscribe to updates from L1 contract")
 		return err
-
-		// TODO(oclaw) metrics
 	}
 	defer sub.Unsubscribe()
 
@@ -181,7 +182,7 @@ func (el *EventListener) subscriber(ctx context.Context, eventCh chan<- *L1Messa
 				Err(err).
 				Msg("L1 subscription is broken")
 
-			// TODO(oclaw) metrics
+			el.metrics.AddSubscriptionError(ctx)
 			return fmt.Errorf("%w: %w", ErrSubscriptionIsBroken, err)
 		}
 		el.logger.Debug().Msg("subscription channel is closed")
@@ -197,6 +198,9 @@ func (el *EventListener) fetcher(ctx context.Context, eventCh chan<- *L1MessageS
 
 	el.logger.Info().Msg("started fetcher")
 
+	el.metrics.SetFetcherActive(ctx)
+	defer el.metrics.SetFetcherIdle(ctx)
+
 	// try fetching as long as possible, force exit after large enough attempt number
 	retrier := common.NewRetryRunner(
 		common.RetryConfig{
@@ -210,12 +214,9 @@ func (el *EventListener) fetcher(ctx context.Context, eventCh chan<- *L1MessageS
 		err := el.fetchPastEvents(ctx, eventCh)
 		if err != nil {
 			el.logger.Error().Err(err).Msg("historical event fetching failed")
-			// TODO (oclaw) metrics
 		}
 		return err
 	})
-
-	// TODO(oclaw) metrics (this routine is not expected to run for too long, we should now if something is stuck here)
 }
 
 func (el *EventListener) fetchPastEvents(ctx context.Context, eventCh chan<- *L1MessageSent) error {
@@ -286,6 +287,7 @@ func (el *EventListener) eventProcessor(
 		if err := el.processEvent(ctx, event); err != nil {
 			return err
 		}
+		el.metrics.AddEventFromFetcher(ctx)
 		processedOldEvents++
 	}
 
@@ -304,6 +306,7 @@ func (el *EventListener) eventProcessor(
 			if err := el.processEvent(ctx, event); err != nil {
 				return err
 			}
+			el.metrics.AddEventFromSubscriber(ctx)
 		case <-ctx.Done():
 			return ctx.Err()
 		}

nil/services/relayer/internal/l1/event_listener_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/NilFoundation/nil/nil/common/logging"
 	"github.com/NilFoundation/nil/nil/internal/db"
+	"github.com/NilFoundation/nil/nil/services/relayer/internal/storage"
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
@@ -21,13 +22,15 @@ type EventListenerTestSuite struct {
 	suite.Suite
 
 	// high level dependencies
-	database db.DB
-	storage  *EventStorage
-	logger   logging.Logger
-	clock    clockwork.Clock
+	database       db.DB
+	storage        *EventStorage
+	storageMetrics storage.TableMetrics
+	logger         logging.Logger
+	clock          clockwork.Clock
 
 	// testing entity
-	listener *EventListener
+	listener        *EventListener
+	listenerMetrics EventListenerMetrics
 
 	// mocks
 	ethClientMock  *EthClientMock
@@ -57,19 +60,32 @@ func (s *EventListenerTestSuite) SetupTest() {
 	s.database, err = db.NewBadgerDbInMemory()
 	s.Require().NoError(err, "failed to initialize test db")
 
+	s.listenerMetrics, err = NewEventListenerMetrics()
+	s.Require().NoError(err)
+
+	s.storageMetrics, err = storage.NewTableMetrics()
+	s.Require().NoError(err)
+
 	s.clock = clockwork.NewRealClock()
 	s.ethClientMock = &EthClientMock{}
 	s.l1ContractMock = &L1ContractMock{}
-
-	s.storage, err = NewEventStorage(s.ctx, s.database, s.clock, nil, s.logger)
+	s.storage, err = NewEventStorage(s.ctx, s.database, s.clock, s.storageMetrics, s.logger)
 	s.Require().NoError(err, "failed to initialize event storage")
 
 	cfg := DefaultEventListenerConfig()
 	cfg.PollInterval = time.Millisecond
 	cfg.BridgeMessengerContractAddress = "0xDEADBEEF"
 	cfg.EmitEventCapacity = 100 // do avoid event dropping
 
-	s.listener, err = NewEventListener(cfg, s.clock, s.ethClientMock, s.l1ContractMock, s.storage, s.logger)
+	s.listener, err = NewEventListener(
+		cfg,
+		s.clock,
+		s.ethClientMock,
+		s.l1ContractMock,
+		s.storage,
+		s.listenerMetrics,
+		s.logger,
+	)
 	s.Require().NoError(err, "failed to create listener")
 }
 

nil/services/relayer/internal/l1/finality_ensurer.go

@@ -76,6 +76,7 @@ type FinalityEnsurer struct {
 	clock         clockwork.Clock
 	l1Storage     *EventStorage
 	l2Storage     *l2.EventStorage
+	metrics       FinalityEnsurerMetrics
 	eventProvider eventProvider
 
 	emitter chan struct{}
@@ -88,6 +89,7 @@ func NewFinalityEnsurer(
 	logger logging.Logger,
 	l1Storage *EventStorage,
 	l2Storage *l2.EventStorage,
+	metrics FinalityEnsurerMetrics,
 	eventProvider eventProvider,
 ) (*FinalityEnsurer, error) {
 	err := config.Validate()
@@ -103,6 +105,7 @@ func NewFinalityEnsurer(
 		l1Storage:     l1Storage,
 		l2Storage:     l2Storage,
 		eventProvider: eventProvider,
+		metrics:       metrics,
 		emitter:       make(chan struct{}, config.EventEmitterCapacity),
 	}
 
@@ -165,6 +168,12 @@ func (fe *FinalityEnsurer) blockFetcher(ctx context.Context) error {
 				if fe.finalizedBlock != nil {
 					log = log.Uint64("local_finalized_block_number", fe.finalizedBlock.BlockNumber)
 				}
+
+				fe.metrics.SetTimeSinceFinalizedBlockNumberUpdate(
+					ctx,
+					uint64(now.Sub(lastSuccessfulUpdate).Seconds()),
+				)
+
 				log.Msg("failed to fetch last finalized block number from Etherium")
 				continue
 			}
@@ -184,8 +193,7 @@ func (fe *FinalityEnsurer) blockFetcher(ctx context.Context) error {
 				Msg("refreshed actual finalized block number")
 			lastSuccessfulUpdate = now
 
-			// TODO(oclaw) metrics
-			// we should alert somehow if we are unable to obtain actual finalized block number long enough
+			fe.metrics.SetTimeSinceFinalizedBlockNumberUpdate(ctx, 0)
 		}
 	}
 }
@@ -205,6 +213,7 @@ func (fe *FinalityEnsurer) pendingEventPoller(ctx context.Context) error {
 		}
 		if err := fe.forwardFinalizedEvents(ctx); err != nil {
 			fe.logger.Error().Err(err).Msg("failed to process l1 pending events")
+			fe.metrics.AddRelayError(ctx)
 		}
 	}
 }
@@ -265,6 +274,11 @@ func (fe *FinalityEnsurer) forwardFinalizedEvents(ctx context.Context) error {
 		Int("orphaned_blocks_count", len(orphaned)).
 		Msg("checked blocks finality")
 
+	var (
+		finalizedEventCount int
+		orphanedEventCount  = len(events)
+	)
+
 	if len(finalized) > 0 {
 		var l2Events []*l2.Event
 		for _, finblk := range finalized {
@@ -277,6 +291,9 @@ func (fe *FinalityEnsurer) forwardFinalizedEvents(ctx context.Context) error {
 			Int("finalized_event_count", len(l2Events)).
 			Msg("saving messages to L2 event storage")
 
+		finalizedEventCount = len(l2Events)
+		orphanedEventCount = len(events) - finalizedEventCount
+
 		err := fe.l2Storage.StoreEvents(ctx, l2Events)
 		if ignoreErrors(err, storage.ErrKeyExists) != nil {
 			return fmt.Errorf("failed to forward events to L2 storage: %w", err)
@@ -289,6 +306,9 @@ func (fe *FinalityEnsurer) forwardFinalizedEvents(ctx context.Context) error {
 		}
 	}
 
+	fe.metrics.AddFinalizedEvents(ctx, uint64(finalizedEventCount))
+	fe.metrics.AddOrphanedEvents(ctx, uint64(orphanedEventCount))
+
 	fe.logger.Info().
 		Int("dropping_events_count", len(events)).
 		Msg("dropping pending events from L1 storage")
@@ -302,7 +322,6 @@ func (fe *FinalityEnsurer) forwardFinalizedEvents(ctx context.Context) error {
 		return fmt.Errorf("failed to cleanup events from l1 storage: %w", err)
 	}
 
-	// TODO(oclaw) metrics
 	return nil
 }
 

nil/services/relayer/internal/l1/finality_ensurer_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/NilFoundation/nil/nil/common/logging"
 	"github.com/NilFoundation/nil/nil/internal/db"
 	"github.com/NilFoundation/nil/nil/services/relayer/internal/l2"
+	"github.com/NilFoundation/nil/nil/services/relayer/internal/storage"
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -50,13 +51,15 @@ type FinalityEnsurerTestSuite struct {
 	suite.Suite
 
 	// high level dependencies
-	database  db.DB
-	l1Storage *EventStorage
-	l2Storage *l2.EventStorage
-	logger    logging.Logger
+	database       db.DB
+	l1Storage      *EventStorage
+	l2Storage      *l2.EventStorage
+	storageMetrics storage.TableMetrics
+	logger         logging.Logger
 
 	// testing entity
-	ensurer *FinalityEnsurer
+	ensurer        *FinalityEnsurer
+	ensurerMetrics FinalityEnsurerMetrics
 
 	// mocks
 	ethClientMock     *EthClientMock
@@ -86,6 +89,9 @@ func (s *FinalityEnsurerTestSuite) SetupTest() {
 	s.database, err = db.NewBadgerDbInMemory()
 	s.Require().NoError(err, "failed to initialize database")
 
+	s.storageMetrics, err = storage.NewTableMetrics()
+	s.Require().NoError(err)
+
 	s.clockMock = clockwork.NewFakeClock()
 
 	s.ethClientMock = &EthClientMock{}
@@ -104,23 +110,27 @@ func (s *FinalityEnsurerTestSuite) SetupTest() {
 		return nil, nil
 	}
 
-	s.l1Storage, err = NewEventStorage(s.ctx, s.database, s.clockMock, nil, s.logger)
+	s.l1Storage, err = NewEventStorage(s.ctx, s.database, s.clockMock, s.storageMetrics, s.logger)
 	s.Require().NoError(err, "failed to initialize L1 storage")
 
-	s.l2Storage = l2.NewEventStorage(s.ctx, s.database, s.clockMock, nil, s.logger)
+	s.l2Storage = l2.NewEventStorage(s.ctx, s.database, s.clockMock, s.storageMetrics, s.logger)
 
 	cfg := DefaultFinalityEnsurerConfig()
 	cfg.EventEmitterCapacity = 100
 
 	s.eventListenerStub = newEventListenerStub()
 
+	s.ensurerMetrics, err = NewFinalityEnsurerMetrics()
+	s.Require().NoError(err)
+
 	s.ensurer, err = NewFinalityEnsurer(
 		cfg,
 		s.ethClientMock,
 		s.clockMock,
 		s.logger,
 		s.l1Storage,
 		s.l2Storage,
+		s.ensurerMetrics,
 		s.eventListenerStub,
 	)
 	s.Require().NoError(err)