Add prometheus metrics for monitoring (#110)

Author: aditya1702

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

go.sum

@@ -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.go

@@ -8,7 +8,9 @@ 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/mock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -20,8 +22,14 @@ func TestAccountModelInsert(t *testing.T) {
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
 
+	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("ObserveDBQueryDuration", "INSERT", "accounts", mock.Anything).Return()
+	mockMetricsService.On("IncDBQuery", "INSERT", "accounts").Return()
+	defer mockMetricsService.AssertExpectations(t)
+
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: mockMetricsService,
 	}
 
 	ctx := context.Background()
@@ -45,8 +53,14 @@ func TestAccountModelDelete(t *testing.T) {
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
 
+	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("ObserveDBQueryDuration", "DELETE", "accounts", mock.Anything).Return()
+	mockMetricsService.On("IncDBQuery", "DELETE", "accounts").Return()
+	defer mockMetricsService.AssertExpectations(t)
+
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: mockMetricsService,
 	}
 
 	ctx := context.Background()
@@ -73,8 +87,14 @@ func TestAccountModelIsAccountFeeBumpEligible(t *testing.T) {
 	require.NoError(t, err)
 	defer dbConnectionPool.Close()
 
+	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("IncDBQuery", "SELECT", "accounts").Return()
+	mockMetricsService.On("ObserveDBQueryDuration", "SELECT", "accounts", mock.Anything).Return()
+	defer mockMetricsService.AssertExpectations(t)
+
 	m := &AccountModel{
-		DB: dbConnectionPool,
+		DB:             dbConnectionPool,
+		MetricsService: mockMetricsService,
 	}
 
 	ctx := context.Background()

internal/data/accounts_test.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/models.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,11 @@ 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)
+	m.MetricsService.IncDBQuery("SELECT", "ingest_store")
 	// First run, key does not exist yet
 	if err == sql.ErrNoRows {
 		return 0, nil
@@ -53,10 +59,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 +100,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,16 +155,19 @@ 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)
 	}
+	m.MetricsService.IncDBQuery("SELECT", "ingest_payments")
 
 	prevExists, nextExists, err := m.existsPrevNext(ctx, filteredSetCTE, address, sort, payments)
 	if err != nil {
 		return nil, false, false, fmt.Errorf("checking prev and next pages: %w", err)
 	}
-
 	return payments, prevExists, nextExists, nil
 }
 
@@ -187,11 +203,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
 }