Port generic chain-agnostic balance monitor to cl-common (#1728)

* Move generic chain-agnostic balance monitor to cl-common

* gomods tidy

* lint

* lint

* fix typo

Author: ogtownsend

pkg/monitoring/balance/generic_balance.go

@@ -0,0 +1,197 @@
+// Package balance provides a generic chain-agnostic balance monitoring service
+// that tracks account balances across different blockchain networks.
+package balance
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/config"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils"
+)
+
+// Config defines the balance monitor configuration.
+type GenericBalanceConfig struct {
+	BalancePollPeriod config.Duration
+}
+
+// GenericBalanceClient defines the interface for getting account balances.
+type GenericBalanceClient interface {
+	GetAccountBalance(addr string) (float64, error)
+}
+
+// GenericBalanceMonitorOpts contains the options for creating a new balance monitor.
+type GenericBalanceMonitorOpts struct {
+	ChainInfo           ChainInfo
+	ChainNativeCurrency string
+
+	Config                  GenericBalanceConfig
+	Logger                  logger.Logger
+	Keystore                core.Keystore
+	NewGenericBalanceClient func() (GenericBalanceClient, error)
+
+	// Maps a public key to an account address (optional, can return key as is)
+	KeyToAccountMapper func(context.Context, string) (string, error)
+}
+
+// ChainInfo contains information about the blockchain network.
+type ChainInfo struct {
+	ChainFamilyName string
+	ChainID         string
+	NetworkName     string
+	NetworkNameFull string
+}
+
+// NewGenericBalanceMonitor returns a balance monitoring services.Service which reports the balance of all Keystore accounts.
+func NewGenericBalanceMonitor(opts GenericBalanceMonitorOpts) (services.Service, error) {
+	// Try to create a new gauge for account balance
+	gauge, err := NewGaugeAccBalance(opts.ChainNativeCurrency)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create gauge: %w", err)
+	}
+
+	lggr := logger.Named(opts.Logger, "BalanceMonitor")
+	return &genericBalanceMonitor{
+		cfg:  opts.Config,
+		lggr: lggr,
+		ks:   opts.Keystore,
+
+		newReader:          opts.NewGenericBalanceClient,
+		keyToAccountMapper: opts.KeyToAccountMapper,
+		updateFn: func(ctx context.Context, acc string, balance float64) {
+			lggr.Infow("Account balance updated", "unit", opts.ChainNativeCurrency, "account", acc, "balance", balance)
+			gauge.Record(ctx, balance, acc, opts.ChainInfo)
+		},
+
+		stop: make(chan struct{}),
+		done: make(chan struct{}),
+	}, nil
+}
+
+type genericBalanceMonitor struct {
+	services.StateMachine
+	cfg  GenericBalanceConfig
+	lggr logger.Logger
+	ks   core.Keystore
+
+	// Returns a new GenericBalanceClient
+	newReader func() (GenericBalanceClient, error)
+	// Maps a public key to an account address (optional, can return key as is)
+	keyToAccountMapper func(context.Context, string) (string, error)
+	// Updates the balance metric
+	updateFn func(ctx context.Context, acc string, balance float64) // overridable for testing
+
+	// Cached instance, intermittently reset to nil.
+	reader GenericBalanceClient
+
+	stop services.StopChan
+	done chan struct{}
+}
+
+func (m *genericBalanceMonitor) Name() string {
+	return m.lggr.Name()
+}
+
+func (m *genericBalanceMonitor) Start(context.Context) error {
+	return m.StartOnce(m.Name(), func() error {
+		go m.start()
+		return nil
+	})
+}
+
+func (m *genericBalanceMonitor) Close() error {
+	return m.StopOnce(m.Name(), func() error {
+		close(m.stop)
+		<-m.done
+		return nil
+	})
+}
+
+func (m *genericBalanceMonitor) HealthReport() map[string]error {
+	return map[string]error{m.Name(): m.Healthy()}
+}
+
+// monitor fn continuously updates balances, until stop signal is received.
+func (m *genericBalanceMonitor) start() {
+	defer close(m.done)
+	ctx, cancel := m.stop.NewCtx()
+	defer cancel()
+
+	period := m.cfg.BalancePollPeriod.Duration()
+	tick := time.After(utils.WithJitter(period))
+	for {
+		select {
+		case <-m.stop:
+			return
+		case <-tick:
+			m.updateBalances(ctx)
+			tick = time.After(utils.WithJitter(period))
+		}
+	}
+}
+
+// getReader returns the stored GenericBalanceClient, creating a new one if necessary.
+func (m *genericBalanceMonitor) getReader() (GenericBalanceClient, error) {
+	if m.reader == nil {
+		var err error
+		m.reader, err = m.newReader()
+		if err != nil {
+			return nil, err
+		}
+	}
+	return m.reader, nil
+}
+
+// updateBalances updates the balances of all accounts in the keystore, using the provided GenericBalanceClient and the updateFn.
+func (m *genericBalanceMonitor) updateBalances(ctx context.Context) {
+	m.lggr.Debug("Updating account balances")
+	keys, err := m.ks.Accounts(ctx)
+	if err != nil {
+		m.lggr.Errorw("Failed to get keys", "err", err)
+		return
+	}
+	if len(keys) == 0 {
+		return
+	}
+	reader, err := m.getReader()
+	if err != nil {
+		m.lggr.Errorw("Failed to get client", "err", err)
+		return
+	}
+
+	var gotSomeBals bool
+	for _, pk := range keys {
+		// Check for shutdown signal, since Balance blocks and may be slow.
+		select {
+		case <-m.stop:
+			return
+		default:
+		}
+
+		// Account address can always be derived from the public key currently
+		// TODO: if we need to support key rotation, the keystore should store the address explicitly
+		// Notice: this is chain-specific key to account mapping injected (e.g., relevant for Aptos key management)
+		accAddr, err := m.keyToAccountMapper(ctx, pk)
+		if err != nil {
+			m.lggr.Errorw("Failed to convert public key to account address", "err", err)
+			continue
+		}
+
+		balance, err := reader.GetAccountBalance(accAddr)
+		if err != nil {
+			m.lggr.Errorw("Failed to get balance", "account", accAddr, "err", err)
+			continue
+		}
+		gotSomeBals = true
+		m.updateFn(ctx, accAddr, balance)
+	}
+
+	// Try a new client next time. // TODO: This is for multinode
+	if !gotSomeBals {
+		m.reader = nil
+	}
+}

pkg/monitoring/balance/metadata.go

@@ -0,0 +1,100 @@
+// Package balance provides a generic chain-agnostic balance monitoring service
+// that tracks account balances across different blockchain networks.
+package balance
+
+import (
+	"encoding/hex"
+
+	"go.opentelemetry.io/otel/attribute"
+)
+
+const (
+	// WorkflowExecutionIDShortLen is the length of the short version of the WorkflowExecutionId (label)
+	WorkflowExecutionIDShortLen = 3 // first 3 characters, 16^3 = 4.096 possibilities (mid-high cardinality)
+)
+
+// TODO: Refactor as a proto referenced from the other proto files (telemetry messages)
+type ExecutionMetadata struct {
+	// Execution Context - Source
+	SourceID string
+	// Execution Context - Chain
+	ChainFamilyName string
+	ChainID         string
+	NetworkName     string
+	NetworkNameFull string
+	// Execution Context - Workflow (capabilities.RequestMetadata)
+	WorkflowID               string
+	WorkflowOwner            string
+	WorkflowExecutionID      string
+	WorkflowName             string
+	WorkflowDonID            uint32
+	WorkflowDonConfigVersion uint32
+	ReferenceID              string
+	// Execution Context - Capability
+	CapabilityType           string
+	CapabilityID             string
+	CapabilityTimestampStart uint32
+	CapabilityTimestampEmit  uint32
+}
+
+// Attributes returns common attributes used for metrics
+func (m ExecutionMetadata) Attributes() []attribute.KeyValue {
+	// Decode workflow name attribute for output
+	workflowName := m.decodeWorkflowName()
+
+	return []attribute.KeyValue{
+		// Execution Context - Source
+		attribute.String("source_id", ValOrUnknown(m.SourceID)),
+		// Execution Context - Chain
+		attribute.String("chain_family_name", ValOrUnknown(m.ChainFamilyName)),
+		attribute.String("chain_id", ValOrUnknown(m.ChainID)),
+		attribute.String("network_name", ValOrUnknown(m.NetworkName)),
+		attribute.String("network_name_full", ValOrUnknown(m.NetworkNameFull)),
+		// Execution Context - Workflow (capabilities.RequestMetadata)
+		attribute.String("workflow_id", ValOrUnknown(m.WorkflowID)),
+		attribute.String("workflow_owner", ValOrUnknown(m.WorkflowOwner)),
+		// Notice: We lower the cardinality on the WorkflowExecutionID so it can be used by metrics
+		// This label has good chances to be unique per workflow, in a reasonable bounded time window
+		// TODO: enable this when sufficiently tested (PromQL queries like alerts might need to change if this is used)
+		// attribute.String("workflow_execution_id_short", ValShortOrUnknown(m.WorkflowExecutionID, WorkflowExecutionIDShortLen)),
+		attribute.String("workflow_name", ValOrUnknown(workflowName)),
+		attribute.Int64("workflow_don_id", int64(m.WorkflowDonID)),
+		attribute.Int64("workflow_don_config_version", int64(m.WorkflowDonConfigVersion)),
+		attribute.String("reference_id", ValOrUnknown(m.ReferenceID)),
+		// Execution Context - Capability
+		attribute.String("capability_type", ValOrUnknown(m.CapabilityType)),
+		attribute.String("capability_id", ValOrUnknown(m.CapabilityID)),
+		// Notice: we don't include the timestamps here (high cardinality)
+	}
+}
+
+// decodeWorkflowName decodes the workflow name from hex string to raw string (underlying, output)
+func (m ExecutionMetadata) decodeWorkflowName() string {
+	bytes, err := hex.DecodeString(m.WorkflowName)
+	if err != nil {
+		// This should never happen
+		bytes = []byte("unknown-decode-error")
+	}
+	return string(bytes)
+}
+
+// ValOrUnknown returns the value if it is not empty, otherwise it returns "unknown"
+// This is needed to avoid issues during exporting OTel metrics to Prometheus
+// For more details see https://smartcontract-it.atlassian.net/browse/INFOPLAT-1349
+func ValOrUnknown(val string) string {
+	if val == "" {
+		return "unknown"
+	}
+	return val
+}
+
+// ValShortOrUnknown returns the short len value if not empty or available, otherwise it returns "unknown"
+func ValShortOrUnknown(val string, maxLen int) string {
+	if val == "" || maxLen <= 0 {
+		return "unknown"
+	}
+	if maxLen > len(val) {
+		return val
+	}
+	return val[:maxLen]
+}

pkg/monitoring/balance/metrics.go

@@ -0,0 +1,50 @@
+// Package balance provides a generic chain-agnostic balance monitoring service
+// that tracks account balances across different blockchain networks.
+package balance
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+)
+
+// GaugeAccBalance defines a new gauge metric for account balance
+type GaugeAccBalance struct {
+	// account_balance
+	gauge metric.Float64Gauge
+}
+
+func NewGaugeAccBalance(unitStr string) (*GaugeAccBalance, error) {
+	name := "account_balance"
+	description := "Balance for configured WT account"
+	gauge, err := beholder.GetMeter().Float64Gauge(name, metric.WithUnit(unitStr), metric.WithDescription(description))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create new gauge %s: %+w", name, err)
+	}
+	return &GaugeAccBalance{gauge}, nil
+}
+
+func (g *GaugeAccBalance) Record(ctx context.Context, balance float64, account string, chainInfo ChainInfo) {
+	oAttrs := metric.WithAttributeSet(g.GetAttributes(account, chainInfo))
+	g.gauge.Record(ctx, balance, oAttrs)
+
+	// TODO: consider also recording record in Prom for availability to NOPs
+}
+
+func (g *GaugeAccBalance) GetAttributes(account string, chainInfo ChainInfo) attribute.Set {
+	return attribute.NewSet(
+		attribute.String("account", account),
+
+		// Execution Context - Source
+		attribute.String("source_id", ValOrUnknown(account)), // reusing account as source_id
+		// Execution Context - Chain
+		attribute.String("chain_family_name", ValOrUnknown(chainInfo.ChainFamilyName)),
+		attribute.String("chain_id", ValOrUnknown(chainInfo.ChainID)),
+		attribute.String("network_name", ValOrUnknown(chainInfo.NetworkName)),
+		attribute.String("network_name_full", ValOrUnknown(chainInfo.NetworkNameFull)),
+	)
+}