feat(executionlayer): metrics (#1426)

fix naming
change reorg log severity

Author: Tangui-Bitfly

backend/pkg/commons/metrics/executionlayer.go

@@ -0,0 +1,38 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	IndexingBlockDifference = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "indexing_block_difference",
+		Help: "Difference between the latest on-chain block and the last indexed block",
+	}, []string{"chainID"})
+
+	IndexingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name: "indexing_time_seconds",
+		Help: "Time taken to index the new blocks",
+	}, []string{"chainID"})
+
+	IndexingTransformerProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name: "indexing_transformer_processing_time_seconds",
+		Help: "Time taken by a transformer to index a block",
+	}, []string{"chainID", "transformer"})
+
+	IndexingPendingBalanceUpdate = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "indexing_pending_balance_update",
+		Help: "Pending balances to update",
+	}, []string{"chainID"})
+
+	IndexingPendingENSUpdate = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "indexing_pending_ens_update",
+		Help: "Pending ENS to update",
+	}, []string{"chainID"})
+
+	IndexingReorgTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "indexing_reorg_total",
+		Help: "Total number of block impacted by chain reorganizations",
+	}, []string{"chainID"})
+)

backend/pkg/executionlayer/blockindexer.go

@@ -31,12 +31,12 @@ type BlockIndexerConfig struct {
 type BlockIndexer struct {
 	store          Store
 	lastBlockStore db2.LastBlocksStore
-	transformers   []TransformFunc
+	transformers   []Transformer
 	client         Client
 	config         BlockIndexerConfig
 }
 
-func NewBlockIndexer(store Store, lastBlockStore db2.LastBlocksStore, config BlockIndexerConfig, client Client, transformers ...TransformFunc) *BlockIndexer {
+func NewBlockIndexer(store Store, lastBlockStore db2.LastBlocksStore, config BlockIndexerConfig, client Client, transformers ...Transformer) *BlockIndexer {
 	return &BlockIndexer{
 		store:          store,
 		lastBlockStore: lastBlockStore,
@@ -170,11 +170,13 @@ func (indexer *BlockIndexer) indexBlock(chainID string, block *types.Eth1Block)
 		Number:  block.Number,
 		Hash:    block.Hash,
 	}
-	for _, transform := range indexer.transformers {
-		err := transform(chainID, block, &res)
+	for _, transformer := range indexer.transformers {
+		start := time.Now()
+		err := transformer.fn(chainID, block, &res)
 		if err != nil {
-			return fmt.Errorf("error transforming block [%v]", block.Number)
+			return fmt.Errorf("error transforming [%v] block [%v]", transformer.name, block.Number)
 		}
+		indexingMetrics.TransformerProcessingTime(chainID, transformer.name, time.Since(start))
 	}
 	if err := indexer.store.AddIndexedBlock(res); err != nil {
 		return fmt.Errorf("error saving block [%v]: %w", block.Number, err)

backend/pkg/executionlayer/blockindexer_test.go

@@ -32,15 +32,15 @@ func TestIndexerWithBigTable(t *testing.T) {
 
 	tests := []struct {
 		name         string
-		transformers []TransformFunc
+		transformers []Transformer
 		action       func(*testing.T) error
 		dataKeys     []string
 		updatesKeys  []string
 		metadataKeys []string
 	}{
 		{
 			name: "transaction",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformTx,
 			},
 			action: func(t *testing.T) error {
@@ -56,7 +56,7 @@ func TestIndexerWithBigTable(t *testing.T) {
 		},
 		{
 			name: "block",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformBlock,
 			},
 			action: func(t *testing.T) error {
@@ -71,7 +71,7 @@ func TestIndexerWithBigTable(t *testing.T) {
 		},
 		{
 			name: "erc20 transfer",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformERC20,
 			},
 			action: func(t *testing.T) error {
@@ -89,7 +89,7 @@ func TestIndexerWithBigTable(t *testing.T) {
 		},
 		{
 			name: "erc721 transfer",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformERC721,
 			},
 			action: func(t *testing.T) error {
@@ -104,7 +104,7 @@ func TestIndexerWithBigTable(t *testing.T) {
 		},
 		{
 			name: "erc1155 transfer",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformERC1155,
 			},
 			action: func(t *testing.T) error {
@@ -119,7 +119,7 @@ func TestIndexerWithBigTable(t *testing.T) {
 		},
 		{
 			name: "contract",
-			transformers: []TransformFunc{
+			transformers: []Transformer{
 				TransformContract,
 			},
 			action: func(t *testing.T) error {

backend/pkg/executionlayer/executionlayer.go

@@ -95,15 +95,25 @@ func (service *IndexerService) SyncRange(start, end uint64, skipNode, skipData b
 
 func (service *IndexerService) SyncLive() {
 	for ; ; time.Sleep(service.config.BlockFrequency) {
+		reorgDepth, err := service.reorgWatcher.LookForReorg()
+		if err != nil {
+			logger.WithField("error", err).Error("reorg lookup")
+			continue
+		}
+		if reorgDepth != 0 {
+			indexingMetrics.ReorgBlockTotal(service.stateReader.chainID, reorgDepth)
+			logger.WithField("depth", reorgDepth).Info("reorg detected")
+		}
+
 		state, err := service.stateReader.state()
 		if err != nil {
 			logger.WithField("error", err).Error("cannot get state")
 			continue
 		}
 		logger.WithFields(state.Fields()).Info("last blocks")
+		indexingMetrics.BlockDifference(state.chainID, state.node-state.LastProcessed())
 
-		if err := service.reorgWatcher.LookForReorg(); err != nil {
-			logger.WithField("error", err).Error("reorg lookup")
+		if state.node == state.LastProcessed() {
 			continue
 		}
 
@@ -184,6 +194,11 @@ type syncState struct {
 	data    uint64
 }
 
+// LastProcessed returns the real last block processed by taking the smallest block between state.data and state.blocks
+func (state syncState) LastProcessed() uint64 {
+	return min(state.data, state.blocks)
+}
+
 func (state syncState) Fields() map[string]interface{} {
 	return map[string]interface{}{
 		"chainID": state.chainID,
@@ -236,8 +251,7 @@ func (service *Indexer) FromHead(state syncState) error {
 	// clear balance cache
 	defer service.balanceCache.Clear(state.chainID)
 
-	// get the real last block processed by taking the smallest block between state.data and state.blocks
-	startBlock := max(min(state.data, state.blocks)+1, 0)
+	startBlock := max(state.LastProcessed()+1, 0)
 	bulk := min(service.config.Bulk, state.node-startBlock+1)
 
 	for ; startBlock <= state.node; startBlock += bulk {
@@ -257,9 +271,11 @@ func (service *Indexer) FromHead(state syncState) error {
 			"elapsed": time.Since(start),
 		}).Info("indexed blocks")
 	}
+	end := time.Since(start) // save end to unify log and metric value
+	indexingMetrics.IndexingTime(state.chainID, end)
 
 	logger.WithFields(logrus.Fields{
-		"duration": time.Since(start),
+		"duration": end,
 	}).Info("indexed head")
 	return nil
 }
@@ -278,6 +294,7 @@ func (service *Indexer) Balances(state syncState) {
 			continue
 		}
 		logger = logger.WithField("pending", total)
+		indexingMetrics.PendingBalanceUpdate(state.chainID, total)
 		if total == 0 {
 			logger.Info("finished updating balances")
 			return
@@ -312,6 +329,7 @@ func (service *Indexer) ENS(state syncState) {
 			continue
 		}
 		logger = logger.WithField("pending", total)
+		indexingMetrics.PendingENSUpdate(state.chainID, total)
 		if total == 0 {
 			logger.Info("finished importing ens")
 			return

backend/pkg/executionlayer/metrics.go

@@ -0,0 +1,35 @@
+package executionlayer
+
+import (
+	"time"
+
+	"github.com/gobitfly/beaconchain/pkg/commons/metrics"
+)
+
+var indexingMetrics = prometheusMetrics{}
+
+type prometheusMetrics struct{}
+
+func (p prometheusMetrics) BlockDifference(chainID string, difference uint64) {
+	metrics.IndexingBlockDifference.WithLabelValues(chainID).Set(float64(difference))
+}
+
+func (p prometheusMetrics) IndexingTime(chainID string, t time.Duration) {
+	metrics.IndexingTime.WithLabelValues(chainID).Observe(t.Seconds())
+}
+
+func (p prometheusMetrics) TransformerProcessingTime(chainID string, transformer string, t time.Duration) {
+	metrics.IndexingTransformerProcessingTime.WithLabelValues(chainID, transformer).Observe(t.Seconds())
+}
+
+func (p prometheusMetrics) PendingBalanceUpdate(chainID string, count int64) {
+	metrics.IndexingPendingBalanceUpdate.WithLabelValues(chainID).Set(float64(count))
+}
+
+func (p prometheusMetrics) PendingENSUpdate(chainID string, count int64) {
+	metrics.IndexingPendingENSUpdate.WithLabelValues(chainID).Set(float64(count))
+}
+
+func (p prometheusMetrics) ReorgBlockTotal(chainID string, depth uint64) {
+	metrics.IndexingReorgTotal.WithLabelValues(chainID).Add(float64(depth))
+}

backend/pkg/executionlayer/reorg.go

@@ -46,10 +46,10 @@ func NewReorgWatcher(client EthClient, store ReorgStore, config ReorgConfig, cha
 	}
 }
 
-func (r *ReorgWatcher) LookForReorg() error {
+func (r *ReorgWatcher) LookForReorg() (uint64, error) {
 	head, err := r.client.HeaderByNumber(context.Background(), nil)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	// ensure we will not try to retrieve blocks that do not exist
@@ -59,32 +59,34 @@ func (r *ReorgWatcher) LookForReorg() error {
 	}
 
 	ctx := context.Background()
-
+	var reorgDepth uint64
 	// for each block check if block node hash and block db hash match
 	for i := head.Number.Uint64() - depth; i <= head.Number.Uint64(); i++ {
 		nodeBlock, err := r.client.HeaderByNumber(ctx, big.NewInt(int64(i)))
 		if err != nil {
-			return err
+			return 0, err
 		}
 		dbBlock, err := r.store.GetBlock(r.chainID, i)
 		if err != nil {
 			// exit if we hit a block that is not yet in the db
 			// it means that we never processed that block or that we revert that block
 			if errors.Is(err, database.ErrNotFound) {
-				return nil
+				break
 			}
-			return err
+			return 0, err
 		}
 
 		if bytes.Equal(nodeBlock.Hash().Bytes(), dbBlock.Hash) {
+			// block was not reorg, continue
 			continue
 		}
+		reorgDepth = head.Number.Uint64() - i + 1
 		log.Warnf("found incosistency at height %v, node block hash: %x, db block hash: %x", i, nodeBlock.Hash().Bytes(), dbBlock.Hash)
 		if err := r.handleReorg(i, head.Number.Uint64()); err != nil {
-			return err
+			return 0, err
 		}
 	}
-	return nil
+	return reorgDepth, nil
 }
 
 func (r *ReorgWatcher) handleReorg(number uint64, head uint64) error {

backend/pkg/executionlayer/reorg_test.go

@@ -80,12 +80,16 @@ func TestReorgWatcher(t *testing.T) {
 			client := newFakeClient(tt.nodeBlocks...)
 			store := newStubReorgStore(tt.dbBlocks...)
 			reorg := NewReorgWatcher(client, store, ReorgConfig{tt.depth}, "chainID", tt.lastBlockStore)
-			if err := reorg.LookForReorg(); err != nil {
+			depth, err := reorg.LookForReorg()
+			if err != nil {
 				t.Fatal(err)
 			}
 			if got, want := store.reverted, tt.reverted; !reflect.DeepEqual(got, want) {
 				t.Errorf("got %v, want %v", got, want)
 			}
+			if got, want := depth, uint64(len(tt.reverted)); got != want {
+				t.Errorf("got %v, want %v", got, want)
+			}
 		})
 	}
 }
@@ -192,7 +196,7 @@ func TestReorgWithBackendAndIndexer(t *testing.T) {
 	// create block that has the same number as revertedBlock (root+1)
 	backend.Commit()
 
-	if err := reorg.LookForReorg(); err != nil {
+	if _, err := reorg.LookForReorg(); err != nil {
 		t.Fatal(err)
 	}