feat(lib/transactions) ready transactions metrics

dot/metrics/collector.go

@@ -0,0 +1,32 @@
+package metrics
+
+import (
+	"time"
+
+	ethmetrics "github.com/ethereum/go-ethereum/metrics"
+)
+
+// GaugeCollector abstracts the Update function to collectors
+// just implement the collection
+type GaugeCollector interface {
+	Update() int64
+}
+
+// CollectGaugeMetrics receives an timeout, label and a gauge collector
+// and acquire the metrics timeout by timeout to a ethereum metrics gauge
+func CollectGaugeMetrics(timeout time.Duration, label string, c GaugeCollector) {
+	t := time.NewTicker(timeout)
+	defer t.Stop()
+
+	collectGauge(label, c)
+
+	for range t.C {
+		collectGauge(label, c)
+	}
+}
+
+func collectGauge(label string, c GaugeCollector) {
+	ethmetrics.Enabled = true
+	pooltx := ethmetrics.GetOrRegisterGauge(label, nil)
+	pooltx.Update(c.Update())
+}

lib/transaction/pool.go

@@ -2,10 +2,15 @@ package transaction
 
 import (
 	"sync"
+	"time"
 
+	"github.com/ChainSafe/gossamer/dot/metrics"
 	"github.com/ChainSafe/gossamer/lib/common"
 )
 
+const collectTxMetricsTimeout = time.Second * 5
+const readyTransactionsMetrics = "gossamer/ready/transaction/metrics"
+
 // Pool represents the transaction pool
 type Pool struct {
 	transactions map[common.Hash]*ValidTransaction
@@ -14,9 +19,17 @@ type Pool struct {
 
 // NewPool returns a new empty Pool
 func NewPool() *Pool {
-	return &Pool{
+	p := &Pool{
 		transactions: make(map[common.Hash]*ValidTransaction),
 	}
+
+	go metrics.CollectGaugeMetrics(
+		collectTxMetricsTimeout,
+		readyTransactionsMetrics,
+		p,
+	)
+
+	return p
 }
 
 // Transactions returns all the transactions in the pool
@@ -49,3 +62,6 @@ func (p *Pool) Remove(hash common.Hash) {
 	defer p.mu.Unlock()
 	delete(p.transactions, hash)
 }
+
+// Update returns the total of valid transactions in the pool
+func (p *Pool) Update() int64 { return int64(len(p.transactions)) }

lib/transaction/pool_test.go

@@ -3,9 +3,12 @@ package transaction
 import (
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/stretchr/testify/require"
+
+	ethmetrics "github.com/ethereum/go-ethereum/metrics"
 )
 
 func TestPool(t *testing.T) {
@@ -50,3 +53,50 @@ func TestPool(t *testing.T) {
 	}
 	require.Equal(t, 0, len(p.Transactions()))
 }
+
+func TestPoolCollectMetrics(t *testing.T) {
+	//reset metric
+	ethmetrics.Unregister(readyTransactionsMetrics)
+
+	ethmetrics.Enabled = true
+	txmetrics := ethmetrics.GetOrRegisterGauge(readyTransactionsMetrics, nil)
+
+	require.Equal(t, int64(0), txmetrics.Value())
+
+	validtx := []*ValidTransaction{
+		{
+			Extrinsic: []byte("a"),
+			Validity:  &Validity{Priority: 1},
+		},
+		{
+			Extrinsic: []byte("b"),
+			Validity:  &Validity{Priority: 4},
+		},
+		{
+			Extrinsic: []byte("c"),
+			Validity:  &Validity{Priority: 2},
+		},
+		{
+			Extrinsic: []byte("d"),
+			Validity:  &Validity{Priority: 17},
+		},
+		{
+			Extrinsic: []byte("e"),
+			Validity:  &Validity{Priority: 2},
+		},
+	}
+
+	h := make([]common.Hash, len(validtx))
+	p := NewPool()
+	for i, v := range validtx {
+		h[i] = p.Insert(v)
+	}
+
+	time.Sleep(collectTxMetricsTimeout + time.Second)
+	require.Equal(t, int64(len(validtx)), txmetrics.Value())
+
+	p.Remove(h[0])
+
+	time.Sleep(collectTxMetricsTimeout + time.Second)
+	require.Equal(t, int64(len(validtx)-1), txmetrics.Value())
+}

lib/transaction/priority_queue.go

@@ -21,10 +21,13 @@ import (
 	"errors"
 	"sync"
 
+	"github.com/ChainSafe/gossamer/dot/metrics"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
 )
 
+const readyPriorityQueueTransactions = "gossamer/ready/transaction/metrics"
+
 // ErrTransactionExists is returned when trying to add a transaction to the queue that already exists
 var ErrTransactionExists = errors.New("transaction is already in queue")
 
@@ -94,6 +97,13 @@ func NewPriorityQueue() *PriorityQueue {
 		pq:  make(priorityQueue, 0),
 		txs: make(map[common.Hash]*Item),
 	}
+
+	go metrics.CollectGaugeMetrics(
+		collectTxMetricsTimeout,
+		readyPriorityQueueTransactions,
+		spq,
+	)
+
 	heap.Init(&spq.pq)
 	return spq
 }
@@ -171,3 +181,6 @@ func (spq *PriorityQueue) Pending() []*ValidTransaction {
 	}
 	return txns
 }
+
+// Update returns the total of valid transactions in the priority queue
+func (spq *PriorityQueue) Update() int64 { return int64(spq.pq.Len()) }

lib/transaction/priority_queue_test.go

@@ -19,6 +19,10 @@ package transaction
 import (
 	"reflect"
 	"testing"
+	"time"
+
+	ethmetrics "github.com/ethereum/go-ethereum/metrics"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPriorityQueue(t *testing.T) {
@@ -45,13 +49,24 @@ func TestPriorityQueue(t *testing.T) {
 		},
 	}
 
+	//check metrics
+	ethmetrics.Unregister(readyPriorityQueueTransactions)
+	ethmetrics.Enabled = true
+
+	priorityQueueM := ethmetrics.GetOrRegisterGauge(readyPriorityQueueTransactions, nil)
+	require.Equal(t, int64(0), priorityQueueM.Value())
+
 	pq := NewPriorityQueue()
 	expected := []int{3, 1, 2, 4, 0}
 
 	for _, node := range tests {
 		pq.Push(node)
 	}
 
+	// wait for metrics to be collected
+	time.Sleep(collectTxMetricsTimeout + time.Second)
+	require.Equal(t, int64(len(tests)), priorityQueueM.Value())
+
 	for i, exp := range expected {
 		n := pq.Pop()
 		if !reflect.DeepEqual(n, tests[exp]) {
@@ -60,6 +75,10 @@ func TestPriorityQueue(t *testing.T) {
 			t.Fatalf("Fail: iteration %d got %v expected %v", i, n, tests[exp])
 		}
 	}
+
+	// wait for metrics to be collected
+	time.Sleep(collectTxMetricsTimeout + time.Second)
+	require.Equal(t, int64(pq.pq.Len()), priorityQueueM.Value())
 }
 
 func TestPriorityQueueAgain(t *testing.T) {