node balance monitoring

cmd/monitoring/main.go

@@ -67,11 +67,15 @@ func main() {
 	)
 	monitor.SourceFactories = append(monitor.SourceFactories, feedBalancesSourceFactory, nodeBalancesSourceFactory)
 
-	promExporterFactory := exporter.NewFeedBalancesFactory(
+	feedBalancesExporterFactory := exporter.NewFeedBalancesFactory(
 		logger.With(log, "component", "solana-prom-exporter"),
 		metrics.NewFeedBalances(logger.With(log, "component", "solana-metrics")),
 	)
-	monitor.ExporterFactories = append(monitor.ExporterFactories, promExporterFactory)
+	nodeBalancesExporterFactory := exporter.NewNodeBalancesFactory(
+		logger.With(log, "component", "solana-prom-exporter"),
+		metrics.NewNodeBalances,
+	)
+	monitor.ExporterFactories = append(monitor.ExporterFactories, feedBalancesExporterFactory, nodeBalancesExporterFactory)
 
 	monitor.Run()
 	log.Infow("monitor stopped")

pkg/monitoring/exporter/feedbalances.go

@@ -40,7 +40,7 @@ func (p *feedBalancesFactory) NewExporter(
 }
 
 func (p *feedBalancesFactory) GetType() string {
-	return "balances"
+	return types.BalanceType
 }
 
 type feeBalances struct {

pkg/monitoring/exporter/nodebalances.go

@@ -0,0 +1,79 @@
+package exporter
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/gagliardetto/solana-go"
+	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/metrics"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+type metricsBuilder func(commonMonitoring.Logger, string) metrics.NodeBalances
+
+func NewNodeBalancesFactory(log commonMonitoring.Logger, metricsFunc metricsBuilder) commonMonitoring.ExporterFactory {
+	return &nodeBalancesFactory{
+		log,
+		metricsFunc,
+	}
+
+}
+
+type nodeBalancesFactory struct {
+	log         commonMonitoring.Logger
+	metricsFunc metricsBuilder
+}
+
+func (f *nodeBalancesFactory) NewExporter(params commonMonitoring.ExporterParams) (commonMonitoring.Exporter, error) {
+	if f.metricsFunc == nil {
+		return nil, fmt.Errorf("metrics generator is nil")
+	}
+	return &nodeBalances{
+		log:     f.log,
+		metrics: metrics.NewNodeBalances(f.log, params.ChainConfig.GetNetworkName()),
+	}, nil
+}
+
+func (f *nodeBalancesFactory) GetType() string {
+	return commonMonitoring.NodesOnlyType(types.BalanceType)
+}
+
+type nodeBalances struct {
+	log     commonMonitoring.Logger
+	metrics metrics.NodeBalances
+
+	lock      sync.Mutex
+	addresses map[string]solana.PublicKey
+}
+
+func (nb *nodeBalances) Export(ctx context.Context, data interface{}) {
+	balances, isBalances := data.(types.Balances)
+	if !isBalances {
+		return
+	}
+	for operator, address := range balances.Addresses {
+		balance, ok := balances.Values[operator]
+		if !ok {
+			nb.log.Errorw("mismatch addresses and balances",
+				"operator", operator,
+				"address", address,
+			)
+			continue
+		}
+		nb.metrics.SetBalance(balance, address.String(), operator)
+	}
+
+	nb.lock.Lock()
+	defer nb.lock.Unlock()
+	nb.addresses = balances.Addresses
+}
+
+func (nb *nodeBalances) Cleanup(_ context.Context) {
+	nb.lock.Lock()
+	defer nb.lock.Unlock()
+	for operator, address := range nb.addresses {
+		nb.metrics.Cleanup(address.String(), operator)
+	}
+}

pkg/monitoring/exporter/nodebalances_test.go

@@ -0,0 +1,48 @@
+package exporter
+
+import (
+	"testing"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/metrics"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/testutils"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+func TestNodeBalances(t *testing.T) {
+	ctx := utils.Context(t)
+	lgr, logs := logger.TestObserved(t, zapcore.ErrorLevel)
+	factory := NewNodeBalancesFactory(lgr, metrics.NewNodeBalances)
+	assert.True(t, commonMonitoring.IsNodesOnly(factory.GetType()))
+
+	chainConfig := testutils.GenerateChainConfig()
+	feedConfig := testutils.GenerateFeedConfig()
+	exporter, err := factory.NewExporter(commonMonitoring.ExporterParams{ChainConfig: chainConfig, FeedConfig: feedConfig, Nodes: []commonMonitoring.NodeConfig{}})
+	require.NoError(t, err)
+
+	// happy path
+	exporter.Export(ctx, types.Balances{
+		Addresses: map[string]solana.PublicKey{t.Name(): {}},
+		Values:    map[string]uint64{t.Name(): 0},
+	})
+
+	exporter.Cleanup(ctx)
+
+	// not balance type
+	assert.NotPanics(t, func() { exporter.Export(ctx, 1) })
+
+	// mismatch data
+	exporter.Export(ctx, types.Balances{
+		Addresses: map[string]solana.PublicKey{t.Name(): {}},
+		Values:    map[string]uint64{},
+	})
+	tests.AssertLogEventually(t, logs, "mismatch addresses and balances")
+}

pkg/monitoring/metrics/metrics.go

@@ -9,22 +9,30 @@ import (
 	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
 )
 
-var feedBalanceLabelNames = []string{
-	// This is the address of the account associated with one of the account names above.
-	"account_address",
-	"feed_id",
-	"chain_id",
-	"contract_status",
-	"contract_type",
-	"feed_name",
-	"feed_path",
-	"network_id",
-	"network_name",
-}
+var (
+	feedBalanceLabelNames = []string{
+		// This is the address of the account associated with one of the account names above.
+		"account_address",
+		"feed_id",
+		"chain_id",
+		"contract_status",
+		"contract_type",
+		"feed_name",
+		"feed_path",
+		"network_id",
+		"network_name",
+	}
+
+	nodeBalanceLabels = []string{
+		"account_address",
+		"node_operator",
+		"chain",
+	}
+)
 
 var gauges map[string]*prometheus.GaugeVec
 
-func makeMetricName(balanceAccountName string) string {
+func makeBalanceMetricName(balanceAccountName string) string {
 	return fmt.Sprintf("sol_balance_%s", balanceAccountName)
 }
 
@@ -35,9 +43,17 @@ func init() {
 	for _, balanceAccountName := range types.FeedBalanceAccountNames {
 		gauges[balanceAccountName] = promauto.NewGaugeVec(
 			prometheus.GaugeOpts{
-				Name: makeMetricName(balanceAccountName),
+				Name: makeBalanceMetricName(balanceAccountName),
 			},
 			feedBalanceLabelNames,
 		)
 	}
+
+	// init gauge for CL node balances
+	gauges[types.NodeBalanceMetric] = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: makeBalanceMetricName(types.NodeBalanceMetric),
+		},
+		nodeBalanceLabels,
+	)
 }

pkg/monitoring/metrics/nodebalances.go

@@ -0,0 +1,52 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+//go:generate mockery --name NodeBalances --output ./mocks/
+
+type NodeBalances interface {
+	SetBalance(balance uint64, address, operator string)
+	Cleanup(address, operator string)
+}
+
+type nodeBalances struct {
+	log   commonMonitoring.Logger
+	chain string
+}
+
+func NewNodeBalances(log commonMonitoring.Logger, chain string) NodeBalances {
+	return &nodeBalances{log, chain}
+}
+
+func (nb *nodeBalances) SetBalance(balance uint64, address, operator string) {
+	gauge, ok := gauges[types.NodeBalanceMetric]
+	if !ok {
+		nb.log.Fatalw("gauge not found", "name", types.NodeBalanceMetric)
+		return
+	}
+
+	gauge.With(prometheus.Labels{
+		"account_address": address,
+		"node_operator":   operator,
+		"chain":           nb.chain,
+	}).Set(float64(balance))
+}
+
+func (nb *nodeBalances) Cleanup(address, operator string) {
+	gauge, ok := gauges[types.NodeBalanceMetric]
+	if !ok {
+		nb.log.Fatalw("gauge not found", "name", types.NodeBalanceMetric)
+		return
+	}
+
+	gauge.Delete(prometheus.Labels{
+		"account_address": address,
+		"node_operator":   operator,
+		"chain":           nb.chain,
+	})
+}

pkg/monitoring/metrics/nodebalances_test.go

@@ -0,0 +1,41 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/testutils"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+func TestNodeBalances(t *testing.T) {
+	m := NewNodeBalances(testutils.NewNullLogger(), t.Name())
+
+	// fetching gauges
+	bal, ok := gauges[types.NodeBalanceMetric]
+	require.True(t, ok)
+
+	v := 100
+	addr := solana.PublicKey{1}.String()
+	operator := t.Name() + "-feed"
+	label := prometheus.Labels{
+		"account_address": addr,
+		"node_operator":   operator,
+		"chain":           t.Name(),
+	}
+
+	// set gauge
+	assert.NotPanics(t, func() { m.SetBalance(uint64(v), addr, operator) })
+	promBal := testutil.ToFloat64(bal.With(label))
+	assert.Equal(t, float64(v), promBal)
+
+	// cleanup gauges
+	assert.Equal(t, 1, testutil.CollectAndCount(bal))
+	assert.NotPanics(t, func() { m.Cleanup(addr, operator) })
+	assert.Equal(t, 0, testutil.CollectAndCount(bal))
+}

pkg/monitoring/source_balances_test.go

@@ -28,7 +28,7 @@ func TestFeedBalancesSource(t *testing.T) {
 	ctx := utils.Context(t)
 
 	factory := NewFeedBalancesSourceFactory(cr, lgr)
-	assert.Equal(t, balancesType, factory.GetType())
+	assert.Equal(t, types.BalanceType, factory.GetType())
 
 	// generate source
 	source, err := factory.NewSource(commonMonitoring.SourceParams{})

pkg/monitoring/source_feed_balances.go

@@ -15,8 +15,6 @@ import (
 )
 
 const (
-	balancesType = "balances"
-
 	ErrBalancesSource = "error while fetching balances"
 	ErrGetBalance     = "GetBalance failed"
 	ErrGetBalanceNil  = "GetBalance returned nil"
@@ -53,7 +51,7 @@ func (s *feedBalancesSourceFactory) NewSource(input commonMonitoring.SourceParam
 }
 
 func (s *feedBalancesSourceFactory) GetType() string {
-	return balancesType
+	return types.BalanceType
 }
 
 type feedBalancesSource struct {

pkg/monitoring/source_node_balances.go

@@ -7,6 +7,7 @@ import (
 
 	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
 	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/config"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
 )
 
 func NewNodeBalancesSourceFactory(
@@ -47,5 +48,5 @@ func (s *nodeBalancesSourceFactory) NewSource(input commonMonitoring.SourceParam
 }
 
 func (s *nodeBalancesSourceFactory) GetType() string {
-	return commonMonitoring.NodesOnlyType(balancesType)
+	return commonMonitoring.NodesOnlyType(types.BalanceType)
 }