funder: prometheus metrics (#869)

## Summary of Changes
- Expose prometheus `/metrics` endpoint on funder
- Emit the following metrics:
- `doublezero_funder_build_info` - Build information of the agent
(version, commit, date)
- `doublezero_funder_errors_total` - Number of errors (counter) with
labels of error_type for distinguishing
- `doublezero_funder_account_balance_sol` - Funder account balance in
SOL (gauge) for alerting when funds are low
- Resolves #856
- Related to #742

## Testing Verification
- Update funder e2e test to check errors and balance metrics

Author: snormore

controlplane/funder/cmd/main.go

@@ -5,6 +5,8 @@ import (
 	"flag"
 	"fmt"
 	"log/slog"
+	"net"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
@@ -13,7 +15,9 @@ import (
 	"github.com/gagliardetto/solana-go"
 	solanarpc "github.com/gagliardetto/solana-go/rpc"
 	"github.com/malbeclabs/doublezero/controlplane/funder/internal/funder"
+	"github.com/malbeclabs/doublezero/controlplane/funder/internal/metrics"
 	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/serviceability"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 const (
@@ -31,6 +35,8 @@ var (
 	topUpSOL                = flag.Float64("top-up-sol", defaultTopUpSOL, "the amount of SOL to top up the funder with")
 	verbose                 = flag.Bool("verbose", false, "enable verbose logging")
 	showVersion             = flag.Bool("version", false, "Print the version of the doublezero-agent and exit")
+	metricsEnable           = flag.Bool("metrics-enable", false, "Enable prometheus metrics")
+	metricsAddr             = flag.String("metrics-addr", ":8080", "Address to listen on for prometheus metrics")
 
 	// Set by LDFLAGS
 	version = "dev"
@@ -72,6 +78,23 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Set up prometheus metrics server if enabled.
+	if *metricsEnable {
+		metrics.BuildInfo.WithLabelValues(version, commit, date).Set(1)
+		go func() {
+			listener, err := net.Listen("tcp", *metricsAddr)
+			if err != nil {
+				log.Error("Failed to start prometheus metrics server listener", "error", err)
+				return
+			}
+			log.Info("Prometheus metrics server listening", "address", listener.Addr())
+			http.Handle("/metrics", promhttp.Handler())
+			if err := http.Serve(listener, nil); err != nil {
+				log.Error("Failed to start prometheus metrics server", "error", err)
+			}
+		}()
+	}
+
 	// Check that keypair path exists.
 	if _, err := os.Stat(*keypairPath); os.IsNotExist(err) {
 		log.Error("Funder keypair does not exist", "path", *keypairPath)

controlplane/funder/internal/funder/config.go

@@ -23,7 +23,7 @@ var (
 )
 
 const (
-	defaultWaitForBalanceTimeout      = 20 * time.Second
+	defaultWaitForBalanceTimeout      = 60 * time.Second
 	defaultWaitForBalancePollInterval = 1 * time.Second
 )
 
@@ -40,6 +40,14 @@ type Config struct {
 	WaitForBalancePollInterval time.Duration
 }
 
+func (c *Config) MinBalanceLamports() uint64 {
+	return uint64(c.MinBalanceSOL * float64(solana.LAMPORTS_PER_SOL))
+}
+
+func (c *Config) TopUpLamports() uint64 {
+	return uint64(c.TopUpSOL * float64(solana.LAMPORTS_PER_SOL))
+}
+
 func (c *Config) Validate() error {
 	if c.Logger == nil {
 		return ErrLoggerRequired

controlplane/funder/internal/funder/funder.go

@@ -9,6 +9,7 @@ import (
 	"github.com/gagliardetto/solana-go"
 	"github.com/gagliardetto/solana-go/programs/system"
 	solanarpc "github.com/gagliardetto/solana-go/rpc"
+	"github.com/malbeclabs/doublezero/controlplane/funder/internal/metrics"
 )
 
 type Funder struct {
@@ -27,11 +28,14 @@ func New(cfg Config) (*Funder, error) {
 }
 
 func (f *Funder) Run(ctx context.Context) error {
+	minBalanceLamports := f.cfg.MinBalanceLamports()
+	topUpLamports := f.cfg.TopUpLamports()
+
 	f.log.Info("Starting funder",
 		"interval", f.cfg.Interval,
 		"signer", f.cfg.Signer.PublicKey(),
-		"minBalanceSOL", f.cfg.MinBalanceSOL,
-		"topUpSOL", f.cfg.TopUpSOL,
+		"minBalanceLamports", minBalanceLamports,
+		"topUpLamports", topUpLamports,
 		"serviceabilityProgramID", f.cfg.Serviceability.ProgramID(),
 	)
 
@@ -47,21 +51,26 @@ func (f *Funder) Run(ctx context.Context) error {
 			err := f.cfg.Serviceability.Load(ctx)
 			if err != nil {
 				f.log.Error("Failed to load serviceability state", "error", err)
+				metrics.Errors.WithLabelValues(metrics.ErrorTypeLoadServiceabilityState).Inc()
 				continue
 			}
 
 			// Check balance of funder.
 			balance, err := f.cfg.Solana.GetBalance(ctx, f.cfg.Signer.PublicKey(), solanarpc.CommitmentFinalized)
 			if err != nil {
 				f.log.Error("Failed to get balance", "error", err)
+				metrics.Errors.WithLabelValues(metrics.ErrorTypeGetFunderAccountBalance).Inc()
 				continue
 			}
 			balanceLamports := balance.Value
-			f.log.Debug("Funder balance", "balance", balanceLamports)
+			f.log.Debug("Funder balance", "account", f.cfg.Signer.PublicKey(), "balanceLamports", balanceLamports)
+			balanceSOL := float64(balanceLamports) / float64(solana.LAMPORTS_PER_SOL)
+			metrics.FunderAccountBalanceSOL.WithLabelValues(f.cfg.Signer.PublicKey().String()).Set(balanceSOL)
 
 			// Check that we have enough SOL to top up metrics publishers.
-			if balanceLamports < uint64(f.cfg.TopUpSOL*float64(solana.LAMPORTS_PER_SOL)) {
-				f.log.Error("Funder balance is below minimum", "balance", balanceLamports, "minBalance", f.cfg.TopUpSOL)
+			if balanceLamports < topUpLamports {
+				f.log.Error("Funder balance is below minimum", "balanceLamports", balanceLamports, "minBalanceLamports", minBalanceLamports)
+				metrics.Errors.WithLabelValues(metrics.ErrorTypeFunderAccountBalanceBelowMinimum).Inc()
 				continue
 			}
 
@@ -80,36 +89,40 @@ func (f *Funder) Run(ctx context.Context) error {
 				balance, err := f.cfg.Solana.GetBalance(ctx, metricsPublisherPK, solanarpc.CommitmentFinalized)
 				if err != nil {
 					f.log.Error("Failed to get balance", "error", err)
+					metrics.Errors.WithLabelValues(metrics.ErrorTypeGetMetricsPublisherAccountBalance).Inc()
 					continue
 				}
 				balanceLamports := balance.Value
-				f.log.Debug("Metrics publisher balance", "device", devicePK, "metricsPublisher", metricsPublisherPK, "balance", balanceLamports, "minBalance", f.cfg.MinBalanceSOL)
+				f.log.Debug("Metrics publisher balance", "device", devicePK, "metricsPublisher", metricsPublisherPK, "balanceLamports", balanceLamports, "minBalanceLamports", minBalanceLamports)
 
 				// If balance is below minimum, top it up.
-				if balanceLamports < uint64(f.cfg.MinBalanceSOL*float64(solana.LAMPORTS_PER_SOL)) {
-					f.log.Info("Topping up metrics publisher", "device", devicePK, "metricsPublisher", metricsPublisherPK, "balance", balanceLamports, "topUp", f.cfg.TopUpSOL)
+				if balanceLamports < minBalanceLamports {
+					f.log.Info("Topping up metrics publisher", "device", devicePK, "metricsPublisher", metricsPublisherPK, "balanceLamports", balanceLamports, "topUpLamports", topUpLamports)
 
-					_, err := transferFunds(ctx, f.cfg.Solana, f.cfg.Signer, metricsPublisherPK, uint64(f.cfg.TopUpSOL*float64(solana.LAMPORTS_PER_SOL)), nil)
+					_, err := transferFunds(ctx, f.cfg.Solana, f.cfg.Signer, metricsPublisherPK, topUpLamports, nil)
 					if err != nil {
 						f.log.Error("Failed to transfer SOL", "error", err)
+						metrics.Errors.WithLabelValues(metrics.ErrorTypeTransferFundsToMetricsPublisher).Inc()
 						continue
 					}
 
 					// Wait for the transfer to complete.
-					err = waitForBalance(ctx, f.cfg.Solana, metricsPublisherPK, f.cfg.MinBalanceSOL, f.cfg.WaitForBalanceTimeout, f.cfg.WaitForBalancePollInterval)
+					f.log.Debug("Waiting for balance", "account", metricsPublisherPK, "expected", minBalanceLamports, "current", balanceLamports)
+					err = waitForBalance(ctx, f.cfg.Solana, metricsPublisherPK, minBalanceLamports, f.cfg.WaitForBalanceTimeout, f.cfg.WaitForBalancePollInterval)
 					if err != nil {
 						f.log.Error("Failed to wait for balance", "error", err)
+						metrics.Errors.WithLabelValues(metrics.ErrorTypeWaitForMetricsPublisherBalance).Inc()
 						continue
 					}
 
-					f.log.Info("Transferred SOL to metrics publisher", "device", devicePK, "metricsPublisher", metricsPublisherPK, "amount", f.cfg.TopUpSOL)
+					f.log.Info("Transferred SOL to metrics publisher", "device", devicePK, "metricsPublisher", metricsPublisherPK, "topUpLamports", topUpLamports)
 				}
 			}
 		}
 	}
 }
 
-func waitForBalance(ctx context.Context, client SolanaClient, account solana.PublicKey, minBalanceSOL float64, timeout time.Duration, pollInterval time.Duration) error {
+func waitForBalance(ctx context.Context, client SolanaClient, account solana.PublicKey, minBalanceLamports uint64, timeout time.Duration, pollInterval time.Duration) error {
 	timer := time.NewTimer(timeout)
 	defer timer.Stop()
 
@@ -119,15 +132,15 @@ func waitForBalance(ctx context.Context, client SolanaClient, account solana.Pub
 			return fmt.Errorf("failed to get balance: %w", err)
 		}
 		balanceLamports := balance.Value
-		if balanceLamports >= uint64(minBalanceSOL*float64(solana.LAMPORTS_PER_SOL)) {
+		if balanceLamports >= minBalanceLamports {
 			return nil
 		}
 
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		case <-timer.C:
-			return fmt.Errorf("timeout waiting for balance: account=%s, expected balance=%.2f SOL", account, minBalanceSOL)
+			return fmt.Errorf("timeout waiting for balance: account=%s, expected balance=%d, current balance=%d", account, minBalanceLamports, balanceLamports)
 		case <-time.After(pollInterval):
 			// continue polling
 		}

controlplane/funder/internal/metrics/metrics.go

@@ -0,0 +1,54 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	// Metrics names.
+	MetricNameBuildInfo               = "doublezero_funder_build_info"
+	MetricNameErrors                  = "doublezero_funder_errors_total"
+	MetricNameFunderAccountBalanceSOL = "doublezero_funder_account_balance_sol"
+
+	// Labels.
+	LabelVersion       = "version"
+	LabelCommit        = "commit"
+	LabelDate          = "date"
+	LabelErrorType     = "error_type"
+	LabelFunderAccount = "funder_account"
+
+	// Error types.
+	ErrorTypeLoadServiceabilityState           = "load_serviceability_state"
+	ErrorTypeGetFunderAccountBalance           = "get_funder_account_balance"
+	ErrorTypeFunderAccountBalanceBelowMinimum  = "funder_account_balance_below_minimum"
+	ErrorTypeGetMetricsPublisherAccountBalance = "get_metrics_publisher_account_balance"
+	ErrorTypeTransferFundsToMetricsPublisher   = "transfer_funds_to_metrics_publisher"
+	ErrorTypeWaitForMetricsPublisherBalance    = "wait_for_metrics_publisher_balance"
+)
+
+var (
+	BuildInfo = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: MetricNameBuildInfo,
+			Help: "Build information of the funder agent",
+		},
+		[]string{LabelVersion, LabelCommit, LabelDate},
+	)
+
+	Errors = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: MetricNameErrors,
+			Help: "Number of errors encountered",
+		},
+		[]string{LabelErrorType},
+	)
+
+	FunderAccountBalanceSOL = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: MetricNameFunderAccountBalanceSOL,
+			Help: "The balance of the funder account in SOL",
+		},
+		[]string{LabelFunderAccount},
+	)
+)

e2e/funder_test.go

@@ -3,6 +3,7 @@
 package e2e_test
 
 import (
+	"fmt"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -59,6 +60,24 @@ func TestE2E_Funder(t *testing.T) {
 	require.NoError(t, err)
 	funderPK := funderPrivateKey.PublicKey()
 
+	// Check that the errors metric only contains "funder_account_balance_below_minimum" errors,
+	// which occur on startup while waiting for the manager/funder account to be funded.
+	metricsClient := dn.Funder.GetMetricsClient()
+	require.NoError(t, metricsClient.Fetch(ctx))
+	errors := metricsClient.GetCounterValues("doublezero_funder_errors_total")
+	require.NotNil(t, errors)
+	require.Len(t, errors, 1)
+	require.Equal(t, "funder_account_balance_below_minimum", errors[0].Labels["error_type"])
+	prevFunderAccountBalanceBelowMinimumCount := int(errors[0].Value)
+
+	// Check the funder account balance metric.
+	require.NoError(t, metricsClient.Fetch(ctx))
+	funderBalance := metricsClient.GetGaugeValues("doublezero_funder_account_balance_sol")
+	require.NotNil(t, funderBalance)
+	// The funder account is the manager account, which we fund with 100 SOL during devnet setup.
+	require.Greater(t, funderBalance[0].Value, 50.0)
+	prevFunderBalance := funderBalance[0].Value
+
 	// Add a device onchain with metrics publisher pubkey.
 	log.Debug("==> Creating LA device onchain")
 	laDeviceMetricsPublisherWallet := solana.NewWallet()
@@ -84,20 +103,78 @@ func TestE2E_Funder(t *testing.T) {
 	// Check that the metrics publisher pubkey is eventually funded.
 	requireEventuallyFunded(t, log, rpcClient, nyDeviceMetricsPublisherWallet.PublicKey(), minBalanceSOL, "NY device metrics publisher")
 
+	// Check that the funder account balance is now lower.
+	require.NoError(t, metricsClient.Fetch(ctx))
+	funderBalance = metricsClient.GetGaugeValues("doublezero_funder_account_balance_sol")
+	require.NotNil(t, funderBalance)
+	require.Less(t, funderBalance[0].Value, prevFunderBalance)
+
 	// Drain current balance from the devices onchain.
-	drainFunds(t, rpcClient, laDeviceMetricsPublisherWallet.PrivateKey, funderPK, 0.01)
-	drainFunds(t, rpcClient, nyDeviceMetricsPublisherWallet.PrivateKey, funderPK, 0.01)
+	drainWallet := solana.NewWallet()
+	log.Debug("--> Draining LA device balance", "account", laDeviceMetricsPublisherWallet.PublicKey())
+	drainFunds(t, rpcClient, laDeviceMetricsPublisherWallet.PrivateKey, drainWallet.PublicKey(), 0.01)
+	log.Debug("--> Draining NY device balance", "account", nyDeviceMetricsPublisherWallet.PublicKey())
+	drainFunds(t, rpcClient, nyDeviceMetricsPublisherWallet.PrivateKey, drainWallet.PublicKey(), 0.01)
 
 	// Check that the devices are eventually funded again.
+	beforeFunderBalance := getBalance(t, rpcClient, funderPK)
 	requireEventuallyFunded(t, log, rpcClient, laDeviceMetricsPublisherWallet.PublicKey(), minBalanceSOL, "LA device metrics publisher")
 	requireEventuallyFunded(t, log, rpcClient, nyDeviceMetricsPublisherWallet.PublicKey(), minBalanceSOL, "NY device metrics publisher")
+
+	// Wait for the funder account balance to show the top up.
+	require.Eventually(t, func() bool {
+		funderBalance := getBalance(t, rpcClient, funderPK)
+		return funderBalance <= beforeFunderBalance-2*topUpSOL
+	}, 60*time.Second, 5*time.Second)
+
+	// Drain the funder account balance to near 0.
+	log.Debug("--> Draining funder account balance", "account", funderPK)
+	drainFunds(t, rpcClient, funderPrivateKey, drainWallet.PublicKey(), 0.01)
+
+	// Check that the errors metric for "funder_account_balance_below_minimum" eventually increases,
+	// which occurs when the funder account balance is drained to below the minimum.
+	require.Eventually(t, func() bool {
+		require.NoError(t, metricsClient.Fetch(ctx))
+		errors = metricsClient.GetCounterValues("doublezero_funder_errors_total")
+		require.NotNil(t, errors)
+		require.Len(t, errors, 1)
+		require.Equal(t, "funder_account_balance_below_minimum", errors[0].Labels["error_type"])
+		if int(errors[0].Value) > prevFunderAccountBalanceBelowMinimumCount {
+			return true
+		}
+		log.Debug("--> Waiting for funder account balance below minimum error to increase", "account", funderPK, "prevCount", prevFunderAccountBalanceBelowMinimumCount, "currentCount", int(errors[0].Value))
+		return false
+	}, 60*time.Second, 5*time.Second)
+
+	// Check that the funder account balance gauge metric is now near 0.
+	require.NoError(t, metricsClient.Fetch(ctx))
+	funderBalance = metricsClient.GetGaugeValues("doublezero_funder_account_balance_sol")
+	require.NotNil(t, funderBalance)
+	require.LessOrEqual(t, funderBalance[0].Value, 0.01)
+
+	// Transfer the drained funds back to the funder account.
+	expectedFunderBalance := drainFunds(t, rpcClient, drainWallet.PrivateKey, funderPrivateKey.PublicKey(), 0.01)
+
+	// Check that the funder account balance is eventually back near the previous value.
+	require.Eventually(t, func() bool {
+		require.NoError(t, metricsClient.Fetch(ctx))
+		funderBalance = metricsClient.GetGaugeValues("doublezero_funder_account_balance_sol")
+		require.NotNil(t, funderBalance)
+		if funderBalance[0].Value > expectedFunderBalance-0.01 && funderBalance[0].Value < expectedFunderBalance+0.01 {
+			return true
+		}
+		log.Debug("--> Waiting for funder account balance to be back near previous value", "account", funderPK, "expectedBalance", expectedFunderBalance, "currentBalance", funderBalance[0].Value)
+		return false
+	}, 60*time.Second, 5*time.Second)
 }
 
-func drainFunds(t *testing.T, client *solanarpc.Client, from solana.PrivateKey, to solana.PublicKey, amount float64) {
+func drainFunds(t *testing.T, client *solanarpc.Client, from solana.PrivateKey, to solana.PublicKey, remainingBalanceSOL float64) float64 {
 	t.Helper()
 
 	balanceSOL := getBalance(t, client, from.PublicKey())
-	transferFunds(t, client, from, to, balanceSOL-amount, nil)
+	transferFunds(t, client, from, to, balanceSOL-remainingBalanceSOL, nil)
+
+	return balanceSOL - remainingBalanceSOL
 }
 
 func requireEventuallyFunded(t *testing.T, log *slog.Logger, client *solanarpc.Client, account solana.PublicKey, minBalanceSOL float64, name string) {
@@ -108,7 +185,7 @@ func requireEventuallyFunded(t *testing.T, log *slog.Logger, client *solanarpc.C
 		require.NoError(t, err)
 		balanceSOL := float64(balance.Value) / float64(solana.LAMPORTS_PER_SOL)
 		if balanceSOL < minBalanceSOL {
-			log.Debug("--> Waiting for %s to be funded", "name", name)
+			log.Debug(fmt.Sprintf("--> Waiting for %s to be funded", name), "account", account, "minBalance", minBalanceSOL, "balance", balanceSOL)
 			return false
 		}
 		return true