Add prometheus metrics for monitoring

go.mod

@@ -5,12 +5,15 @@ go 1.23.2
 require (
 	github.com/alitto/pond v1.9.2
 	github.com/aws/aws-sdk-go v1.55.5
+	github.com/dlmiddlecote/sqlstats v1.0.2
 	github.com/getsentry/sentry-go v0.29.1
 	github.com/go-chi/chi v4.1.2+incompatible
 	github.com/go-playground/validator/v10 v10.22.1
 	github.com/google/uuid v1.6.0
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/lib/pq v1.10.9
+	github.com/mattn/go-sqlite3 v1.14.22
+	github.com/prometheus/client_golang v1.17.0
 	github.com/rubenv/sql-migrate v1.7.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.8.1
@@ -49,7 +52,6 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_golang v1.17.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect

internal/data/accounts.go

@@ -3,31 +3,41 @@ package data
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/stellar/wallet-backend/internal/db"
+	"github.com/stellar/wallet-backend/internal/metrics"
 )
 
 type AccountModel struct {
-	DB db.ConnectionPool
+	DB             db.ConnectionPool
+	MetricsService *metrics.MetricsService
 }
 
 func (m *AccountModel) Insert(ctx context.Context, address string) error {
 	const query = `INSERT INTO accounts (stellar_address) VALUES ($1) ON CONFLICT DO NOTHING`
+	start := time.Now()
 	_, err := m.DB.ExecContext(ctx, query, address)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("INSERT", "accounts", duration)
 	if err != nil {
 		return fmt.Errorf("inserting address %s: %w", address, err)
 	}
+	m.MetricsService.IncDBQuery("INSERT", "accounts")
 
 	return nil
 }
 
 func (m *AccountModel) Delete(ctx context.Context, address string) error {
 	const query = `DELETE FROM accounts WHERE stellar_address = $1`
+	start := time.Now()
 	_, err := m.DB.ExecContext(ctx, query, address)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("DELETE", "accounts", duration)
 	if err != nil {
 		return fmt.Errorf("deleting address %s: %w", address, err)
 	}
-
+	m.MetricsService.IncDBQuery("DELETE", "accounts")
 	return nil
 }
 
@@ -43,10 +53,13 @@ func (m *AccountModel) IsAccountFeeBumpEligible(ctx context.Context, address str
 			)
 	`
 	var exists bool
+	start := time.Now()
 	err := m.DB.GetContext(ctx, &exists, query, address)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("SELECT", "accounts", duration)
 	if err != nil {
 		return false, fmt.Errorf("checking if account %s is fee bump eligible: %w", address, err)
 	}
-
+	m.MetricsService.IncDBQuery("SELECT", "accounts")
 	return exists, nil
 }

internal/data/accounts_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stellar/go/keypair"
 	"github.com/stellar/wallet-backend/internal/db"
 	"github.com/stellar/wallet-backend/internal/db/dbtest"
+	"github.com/stellar/wallet-backend/internal/metrics"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -19,9 +20,13 @@ func TestAccountModelInsert(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	ctx := context.Background()
@@ -44,9 +49,13 @@ func TestAccountModelDelete(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	ctx := context.Background()
@@ -72,9 +81,13 @@ func TestAccountModelIsAccountFeeBumpEligible(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	ctx := context.Background()

internal/data/models.go

@@ -4,20 +4,21 @@ import (
 	"errors"
 
 	"github.com/stellar/wallet-backend/internal/db"
+	"github.com/stellar/wallet-backend/internal/metrics"
 )
 
 type Models struct {
 	Payments *PaymentModel
 	Account  *AccountModel
 }
 
-func NewModels(db db.ConnectionPool) (*Models, error) {
+func NewModels(db db.ConnectionPool, metricsService *metrics.MetricsService) (*Models, error) {
 	if db == nil {
 		return nil, errors.New("ConnectionPool must be initialized")
 	}
 
 	return &Models{
-		Payments: &PaymentModel{DB: db},
-		Account:  &AccountModel{DB: db},
+		Payments: &PaymentModel{DB: db, MetricsService: metricsService},
+		Account:  &AccountModel{DB: db, MetricsService: metricsService},
 	}, nil
 }

internal/data/payments.go

@@ -8,10 +8,12 @@ import (
 	"time"
 
 	"github.com/stellar/wallet-backend/internal/db"
+	"github.com/stellar/wallet-backend/internal/metrics"
 )
 
 type PaymentModel struct {
-	DB db.ConnectionPool
+	DB             db.ConnectionPool
+	MetricsService *metrics.MetricsService
 }
 
 type Payment struct {
@@ -36,7 +38,10 @@ type Payment struct {
 
 func (m *PaymentModel) GetLatestLedgerSynced(ctx context.Context, cursorName string) (uint32, error) {
 	var lastSyncedLedger uint32
+	start := time.Now()
 	err := m.DB.GetContext(ctx, &lastSyncedLedger, `SELECT value FROM ingest_store WHERE key = $1`, cursorName)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("SELECT", "ingest_store", duration)
 	// First run, key does not exist yet
 	if err == sql.ErrNoRows {
 		return 0, nil
@@ -53,10 +58,14 @@ func (m *PaymentModel) UpdateLatestLedgerSynced(ctx context.Context, cursorName
 		INSERT INTO ingest_store (key, value) VALUES ($1, $2)
 		ON CONFLICT (key) DO UPDATE SET value = excluded.value
 	`
+	start := time.Now()
 	_, err := m.DB.ExecContext(ctx, query, cursorName, ledger)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("INSERT", "ingest_store", duration)
 	if err != nil {
 		return fmt.Errorf("updating last synced ledger to %d: %w", ledger, err)
 	}
+	m.MetricsService.IncDBQuery("INSERT", "ingest_store")
 
 	return nil
 }
@@ -90,12 +99,15 @@ func (m *PaymentModel) AddPayment(ctx context.Context, tx db.Transaction, paymen
 			memo_type = EXCLUDED.memo_type
 		;
 	`
+	start := time.Now()
 	_, err := tx.ExecContext(ctx, query, payment.OperationID, payment.OperationType, payment.TransactionID, payment.TransactionHash, payment.FromAddress, payment.ToAddress, payment.SrcAssetCode, payment.SrcAssetIssuer, payment.SrcAssetType, payment.SrcAmount,
 		payment.DestAssetCode, payment.DestAssetIssuer, payment.DestAssetType, payment.DestAmount, payment.CreatedAt, payment.Memo, payment.MemoType)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("INSERT", "ingest_payments", duration)
 	if err != nil {
 		return fmt.Errorf("inserting payment: %w", err)
 	}
-
+	m.MetricsService.IncDBQuery("INSERT", "ingest_payments")
 	return nil
 }
 
@@ -142,7 +154,10 @@ func (m *PaymentModel) GetPaymentsPaginated(ctx context.Context, address string,
 	if err != nil {
 		return nil, false, false, fmt.Errorf("preparing named query: %w", err)
 	}
+	start := time.Now()
 	err = m.DB.SelectContext(ctx, &payments, query, args...)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("SELECT", "ingest_payments", duration)
 	if err != nil {
 		return nil, false, false, fmt.Errorf("fetching payments: %w", err)
 	}
@@ -151,7 +166,7 @@ func (m *PaymentModel) GetPaymentsPaginated(ctx context.Context, address string,
 	if err != nil {
 		return nil, false, false, fmt.Errorf("checking prev and next pages: %w", err)
 	}
-
+	m.MetricsService.IncDBQuery("SELECT", "ingest_payments")
 	return payments, prevExists, nextExists, nil
 }
 
@@ -187,11 +202,14 @@ func (m *PaymentModel) existsPrevNext(ctx context.Context, filteredSetCTE string
 	}
 
 	var prevExists, nextExists bool
+	start := time.Now()
 	err = m.DB.QueryRowxContext(ctx, query, args...).Scan(&prevExists, &nextExists)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("SELECT", "ingest_payments", duration)
 	if err != nil {
 		return false, false, fmt.Errorf("fetching prev and next exists: %w", err)
 	}
-
+	m.MetricsService.IncDBQuery("SELECT", "ingest_payments")
 	return prevExists, nextExists, nil
 }
 

internal/data/payments_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stellar/go/xdr"
 	"github.com/stellar/wallet-backend/internal/db"
 	"github.com/stellar/wallet-backend/internal/db/dbtest"
+	"github.com/stellar/wallet-backend/internal/metrics"
 	"github.com/stellar/wallet-backend/internal/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -21,9 +22,13 @@ func TestPaymentModelAddPayment(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	m := &PaymentModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 	ctx := context.Background()
 
@@ -145,10 +150,14 @@ func TestPaymentModelGetLatestLedgerSynced(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	ctx := context.Background()
 	m := &PaymentModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	const key = "ingest_store_key"
@@ -170,10 +179,14 @@ func TestPaymentModelUpdateLatestLedgerSynced(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	ctx := context.Background()
 	m := &PaymentModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	const key = "ingest_store_key"
@@ -192,10 +205,14 @@ func TestPaymentModelGetPaymentsPaginated(t *testing.T) {
 	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
+	sqlxDB, err := dbConnectionPool.SqlxDB(context.Background())
+	require.NoError(t, err)
+	metricsService := metrics.NewMetricsService(sqlxDB)
 
 	ctx := context.Background()
 	m := &PaymentModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: metricsService,
 	}
 
 	dbPayments := []Payment{

internal/ingest/ingest.go

@@ -6,12 +6,14 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
 	"github.com/stellar/go/support/log"
 
 	"github.com/stellar/wallet-backend/internal/apptracker"
 	"github.com/stellar/wallet-backend/internal/data"
 	"github.com/stellar/wallet-backend/internal/db"
+	"github.com/stellar/wallet-backend/internal/metrics"
 	"github.com/stellar/wallet-backend/internal/services"
 	"github.com/stellar/wallet-backend/internal/tss"
 	tssrouter "github.com/stellar/wallet-backend/internal/tss/router"
@@ -49,22 +51,26 @@ func Ingest(cfg Configs) error {
 }
 
 func setupDeps(cfg Configs) (services.IngestService, error) {
-	// Open DB connection pool
 	dbConnectionPool, err := db.OpenDBConnectionPool(cfg.DatabaseURL)
 	if err != nil {
 		return nil, fmt.Errorf("connecting to the database: %w", err)
 	}
-	models, err := data.NewModels(dbConnectionPool)
+	db, err := dbConnectionPool.SqlxDB(context.Background())
+	if err != nil {
+		return nil, fmt.Errorf("getting sqlx db: %w", err)
+	}
+	metricsService := metrics.NewMetricsService(db)
+	models, err := data.NewModels(dbConnectionPool, metricsService)
 	if err != nil {
 		return nil, fmt.Errorf("creating models: %w", err)
 	}
 	httpClient := &http.Client{Timeout: 30 * time.Second}
-	rpcService, err := services.NewRPCService(cfg.RPCURL, httpClient)
+	rpcService, err := services.NewRPCService(cfg.RPCURL, httpClient, metricsService)
 	if err != nil {
 		return nil, fmt.Errorf("instantiating rpc service: %w", err)
 	}
 	go rpcService.TrackRPCServiceHealth(context.Background())
-	tssStore, err := tssstore.NewStore(dbConnectionPool)
+	tssStore, err := tssstore.NewStore(dbConnectionPool, metricsService)
 	if err != nil {
 		return nil, fmt.Errorf("instantiating tss store: %w", err)
 	}
@@ -75,10 +81,18 @@ func setupDeps(cfg Configs) (services.IngestService, error) {
 	router := tssrouter.NewRouter(tssRouterConfig)
 
 	ingestService, err := services.NewIngestService(
-		models, cfg.LedgerCursorName, cfg.AppTracker, rpcService, router, tssStore)
+		models, cfg.LedgerCursorName, cfg.AppTracker, rpcService, router, tssStore, metricsService)
 	if err != nil {
 		return nil, fmt.Errorf("instantiating ingest service: %w", err)
 	}
 
+	http.Handle("/ingest-metrics", promhttp.HandlerFor(metricsService.GetRegistry(), promhttp.HandlerOpts{}))
+	go func() {
+		err := http.ListenAndServe(":8002", nil)
+		if err != nil {
+			log.Ctx(context.Background()).Fatalf("starting ingest metrics server: %v", err)
+		}
+	}()
+
 	return ingestService, nil
 }