Merge pull request #595 from onflow/petera/add-indexer-metrics

Add block/tx indexed metrics and spork client logging

metrics/collector.go

@@ -15,6 +15,7 @@ type Collector interface {
 	ServerPanicked(reason string)
 	CadenceHeightIndexed(height uint64)
 	EVMHeightIndexed(height uint64)
+	EVMTransactionIndexed(count int)
 	EVMAccountInteraction(address string)
 	MeasureRequestDuration(start time.Time, method string)
 	OperatorBalance(account *flow.Account)
@@ -29,6 +30,8 @@ type DefaultCollector struct {
 	serverPanicsCounters      *prometheus.CounterVec
 	cadenceBlockHeight        prometheus.Gauge
 	evmBlockHeight            prometheus.Gauge
+	evmBlockIndexedCounter    prometheus.Counter
+	evmTxIndexedCounter       prometheus.Counter
 	operatorBalance           prometheus.Gauge
 	evmAccountCallCounters    *prometheus.CounterVec
 	requestDurations          *prometheus.HistogramVec
@@ -65,6 +68,16 @@ func NewCollector(logger zerolog.Logger) Collector {
 		Help: "Current EVM block height",
 	})
 
+	evmBlockIndexedCounter := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: prefixedName("blocks_indexed_total"),
+		Help: "Total number of blocks indexed",
+	})
+
+	evmTxIndexedCounter := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: prefixedName("txs_indexed_total"),
+		Help: "Total number transactions indexed",
+	})
+
 	evmAccountCallCounters := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: prefixedName("evm_account_interactions_total"),
 		Help: "Total number of account interactions",
@@ -83,6 +96,8 @@ func NewCollector(logger zerolog.Logger) Collector {
 		serverPanicsCounters,
 		cadenceBlockHeight,
 		evmBlockHeight,
+		evmBlockIndexedCounter,
+		evmTxIndexedCounter,
 		operatorBalance,
 		evmAccountCallCounters,
 		requestDurations,
@@ -98,6 +113,8 @@ func NewCollector(logger zerolog.Logger) Collector {
 		serverPanicsCounters:      serverPanicsCounters,
 		cadenceBlockHeight:        cadenceBlockHeight,
 		evmBlockHeight:            evmBlockHeight,
+		evmBlockIndexedCounter:    evmBlockIndexedCounter,
+		evmTxIndexedCounter:       evmTxIndexedCounter,
 		evmAccountCallCounters:    evmAccountCallCounters,
 		requestDurations:          requestDurations,
 		operatorBalance:           operatorBalance,
@@ -133,6 +150,11 @@ func (c *DefaultCollector) CadenceHeightIndexed(height uint64) {
 
 func (c *DefaultCollector) EVMHeightIndexed(height uint64) {
 	c.evmBlockHeight.Set(float64(height))
+	c.evmBlockIndexedCounter.Inc()
+}
+
+func (c *DefaultCollector) EVMTransactionIndexed(count int) {
+	c.evmTxIndexedCounter.Add(float64(count))
 }
 
 func (c *DefaultCollector) EVMAccountInteraction(address string) {

metrics/nop.go

@@ -17,6 +17,7 @@ func (c *nopCollector) TraceDownloadFailed()                     {}
 func (c *nopCollector) ServerPanicked(string)                    {}
 func (c *nopCollector) CadenceHeightIndexed(uint64)              {}
 func (c *nopCollector) EVMHeightIndexed(uint64)                  {}
+func (c *nopCollector) EVMTransactionIndexed(int)                {}
 func (c *nopCollector) EVMAccountInteraction(string)             {}
 func (c *nopCollector) MeasureRequestDuration(time.Time, string) {}
 func (c *nopCollector) OperatorBalance(*flow.Account)            {}

services/ingestion/engine.go

@@ -197,6 +197,7 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 		}
 	}
 
+	e.collector.EVMTransactionIndexed(len(events.Transactions()))
 	e.collector.EVMHeightIndexed(events.Block().Height)
 	e.collector.CadenceHeightIndexed(events.CadenceHeight())
 	return nil

services/requester/cross-spork_client.go

@@ -28,7 +28,7 @@ func (s *sporkClient) contains(height uint64) bool {
 type sporkClients []*sporkClient
 
 // addSpork will add a new spork host defined by the first and last height boundary in that spork.
-func (s *sporkClients) add(client access.Client) error {
+func (s *sporkClients) add(logger zerolog.Logger, client access.Client) error {
 	header, err := client.GetLatestBlockHeader(context.Background(), true)
 	if err != nil {
 		return fmt.Errorf("could not get latest height using the spork client: %w", err)
@@ -39,6 +39,11 @@ func (s *sporkClients) add(client access.Client) error {
 		return fmt.Errorf("could not get node info using the spork client: %w", err)
 	}
 
+	logger.Info().
+		Uint64("firstHeight", info.NodeRootBlockHeight).
+		Uint64("lastHeight", header.Height).
+		Msg("adding spork client")
+
 	*s = append(*s, &sporkClient{
 		firstHeight: info.NodeRootBlockHeight,
 		lastHeight:  header.Height,
@@ -113,7 +118,7 @@ func NewCrossSporkClient(
 
 	clients := &sporkClients{}
 	for _, c := range pastSporks {
-		if err := clients.add(c); err != nil {
+		if err := clients.add(logger, c); err != nil {
 			return nil, err
 		}
 	}

services/requester/cross-spork_client_test.go

@@ -34,16 +34,17 @@ func Test_CrossSporkClient(t *testing.T) {
 }
 
 func Test_CrossSporkClients(t *testing.T) {
+	logger := zerolog.Nop()
 	t.Run("add and validate", func(t *testing.T) {
 		clients := &sporkClients{}
 
 		client1 := testutils.SetupClientForRange(10, 100)
 		client2 := testutils.SetupClientForRange(101, 200)
 		client3 := testutils.SetupClientForRange(201, 300)
 
-		require.NoError(t, clients.add(client2))
-		require.NoError(t, clients.add(client3))
-		require.NoError(t, clients.add(client1))
+		require.NoError(t, clients.add(logger, client2))
+		require.NoError(t, clients.add(logger, client3))
+		require.NoError(t, clients.add(logger, client1))
 
 		require.True(t, clients.continuous())
 
@@ -58,14 +59,14 @@ func Test_CrossSporkClients(t *testing.T) {
 		require.Equal(t, nil, clients.get(310))
 	})
 
-	t.Run("add and validate not-continues", func(t *testing.T) {
+	t.Run("add and validate not-continuous", func(t *testing.T) {
 		clients := &sporkClients{}
 
 		client1 := testutils.SetupClientForRange(10, 30)
 		client2 := testutils.SetupClientForRange(50, 80)
 
-		require.NoError(t, clients.add(client1))
-		require.NoError(t, clients.add(client2))
+		require.NoError(t, clients.add(logger, client1))
+		require.NoError(t, clients.add(logger, client2))
 
 		require.False(t, clients.continuous())
 	})