[BUG] [internal] Pool metrics fixes

go.mod

@@ -6,7 +6,6 @@ toolchain go1.24.2
 
 require (
 	github.com/99designs/gqlgen v0.17.76
-	github.com/alitto/pond v1.9.2
 	github.com/alitto/pond/v2 v2.5.0
 	github.com/avast/retry-go/v4 v4.6.1
 	github.com/aws/aws-sdk-go v1.55.7

go.sum

@@ -56,8 +56,6 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/alitto/pond v1.9.2 h1:9Qb75z/scEZVCoSU+osVmQ0I0JOeLfdTDafrbcJ8CLs=
-github.com/alitto/pond v1.9.2/go.mod h1:xQn3P/sHTYcU/1BR3i86IGIrilcrGC2LiS+E2+CJWsI=
 github.com/alitto/pond/v2 v2.5.0 h1:vPzS5GnvSDRhWQidmj2djHllOmjFExVFbDGCw1jdqDw=
 github.com/alitto/pond/v2 v2.5.0/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNgfBlViaCIJKLlCJ6/fmUseuG0wVQ=

internal/metrics/metrics.go

@@ -4,14 +4,14 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/alitto/pond"
+	"github.com/alitto/pond/v2"
 	"github.com/dlmiddlecote/sqlstats"
 	"github.com/jmoiron/sqlx"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 type MetricsService interface {
-	RegisterPoolMetrics(channel string, pool *pond.WorkerPool)
+	RegisterPoolMetrics(channel string, pool pond.Pool)
 	GetRegistry() *prometheus.Registry
 	SetLatestLedgerIngested(value float64)
 	ObserveIngestionDuration(ingestionType string, duration float64)
@@ -419,10 +419,10 @@ func (m *metricsService) registerMetrics() {
 }
 
 // RegisterPool registers a worker pool for metrics collection
-func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPool) {
+func (m *metricsService) RegisterPoolMetrics(channel string, pool pond.Pool) {
 	m.registry.MustRegister(prometheus.NewGaugeFunc(
 		prometheus.GaugeOpts{
-			Name:        fmt.Sprintf("pool_workers_running_%s", channel),
+			Name:        "pool_workers_running",
 			Help:        "Number of running worker goroutines",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},
@@ -431,20 +431,9 @@ func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPo
 		},
 	))
 
-	m.registry.MustRegister(prometheus.NewGaugeFunc(
-		prometheus.GaugeOpts{
-			Name:        fmt.Sprintf("pool_workers_idle_%s", channel),
-			Help:        "Number of idle worker goroutines",
-			ConstLabels: prometheus.Labels{"channel": channel},
-		},
-		func() float64 {
-			return float64(pool.IdleWorkers())
-		},
-	))
-
 	m.registry.MustRegister(prometheus.NewCounterFunc(
 		prometheus.CounterOpts{
-			Name:        fmt.Sprintf("pool_tasks_submitted_total_%s", channel),
+			Name:        "pool_tasks_submitted_total",
 			Help:        "Number of tasks submitted",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},
@@ -455,7 +444,7 @@ func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPo
 
 	m.registry.MustRegister(prometheus.NewGaugeFunc(
 		prometheus.GaugeOpts{
-			Name:        fmt.Sprintf("pool_tasks_waiting_%s", channel),
+			Name:        "pool_tasks_waiting",
 			Help:        "Number of tasks currently waiting in the queue",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},
@@ -466,7 +455,7 @@ func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPo
 
 	m.registry.MustRegister(prometheus.NewCounterFunc(
 		prometheus.CounterOpts{
-			Name:        fmt.Sprintf("pool_tasks_successful_total_%s", channel),
+			Name:        "pool_tasks_successful_total",
 			Help:        "Number of tasks that completed successfully",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},
@@ -477,7 +466,7 @@ func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPo
 
 	m.registry.MustRegister(prometheus.NewCounterFunc(
 		prometheus.CounterOpts{
-			Name:        fmt.Sprintf("pool_tasks_failed_total_%s", channel),
+			Name:        "pool_tasks_failed_total",
 			Help:        "Number of tasks that completed with panic",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},
@@ -488,7 +477,7 @@ func (m *metricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPo
 
 	m.registry.MustRegister(prometheus.NewCounterFunc(
 		prometheus.CounterOpts{
-			Name:        fmt.Sprintf("pool_tasks_completed_total_%s", channel),
+			Name:        "pool_tasks_completed_total",
 			Help:        "Number of tasks that completed either successfully or with panic",
 			ConstLabels: prometheus.Labels{"channel": channel},
 		},

internal/metrics/metrics_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/alitto/pond"
+	"github.com/alitto/pond/v2"
 	"github.com/jmoiron/sqlx"
 	_ "github.com/mattn/go-sqlite3" // SQLite driver
 	"github.com/stretchr/testify/assert"
@@ -609,14 +609,12 @@ func TestIngestionPhaseMetrics(t *testing.T) {
 }
 
 func TestPoolMetrics(t *testing.T) {
-	db := setupTestDB(t)
-	defer db.Close()
-
-	ms := NewMetricsService(db)
-
 	t.Run("worker pool metrics - success case", func(t *testing.T) {
+		db := setupTestDB(t)
+		defer db.Close()
+		ms := NewMetricsService(db)
 		channel := "test_channel"
-		pool := pond.New(5, 10)
+		pool := pond.NewPool(5)
 
 		ms.RegisterPoolMetrics(channel, pool)
 
@@ -639,31 +637,27 @@ func TestPoolMetrics(t *testing.T) {
 		for _, mf := range metricFamilies {
 			metric := mf.GetMetric()[0]
 			switch mf.GetName() {
-			case "pool_workers_running_" + channel:
+			case "pool_workers_running":
 				metricValues["workers_running"] = metric.GetGauge().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for workers_running")
 
-			case "pool_workers_idle_" + channel:
-				metricValues["workers_idle"] = metric.GetGauge().GetValue()
-				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for workers_idle")
-
-			case "pool_tasks_submitted_total_" + channel:
+			case "pool_tasks_submitted_total":
 				metricValues["tasks_submitted"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for tasks_submitted")
 
-			case "pool_tasks_completed_total_" + channel:
+			case "pool_tasks_completed_total":
 				metricValues["tasks_completed"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for tasks_completed")
 
-			case "pool_tasks_successful_total_" + channel:
+			case "pool_tasks_successful_total":
 				metricValues["tasks_successful"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for tasks_successful")
 
-			case "pool_tasks_failed_total_" + channel:
+			case "pool_tasks_failed_total":
 				metricValues["tasks_failed"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for tasks_failed")
 
-			case "pool_tasks_waiting_" + channel:
+			case "pool_tasks_waiting":
 				metricValues["tasks_waiting"] = metric.GetGauge().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for tasks_waiting")
 			}
@@ -675,14 +669,16 @@ func TestPoolMetrics(t *testing.T) {
 		assert.Equal(t, float64(0), metricValues["tasks_failed"], "Expected 0 failed tasks")
 		assert.Equal(t, float64(0), metricValues["tasks_waiting"], "Expected 0 waiting tasks")
 
-		// Workers should be idle after tasks complete
-		assert.Equal(t, float64(3), metricValues["workers_idle"], "Should have idle workers")
 		pool.StopAndWait()
 	})
 
 	t.Run("worker pool metrics - with failures", func(t *testing.T) {
+		db := setupTestDB(t)
+		defer db.Close()
+		ms := NewMetricsService(db)
+
 		channel := "test_channel_failures"
-		pool := pond.New(2, 5)
+		pool := pond.NewPool(2)
 
 		ms.RegisterPoolMetrics(channel, pool)
 
@@ -709,16 +705,16 @@ func TestPoolMetrics(t *testing.T) {
 		for _, mf := range metricFamilies {
 			metric := mf.GetMetric()[0]
 			switch mf.GetName() {
-			case "pool_tasks_failed_total_" + channel:
+			case "pool_tasks_failed_total":
 				metricValues["tasks_failed"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for failed tasks")
-			case "pool_tasks_successful_total_" + channel:
+			case "pool_tasks_successful_total":
 				metricValues["tasks_successful"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for successful tasks")
-			case "pool_tasks_submitted_total_" + channel:
+			case "pool_tasks_submitted_total":
 				metricValues["tasks_submitted"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for submitted tasks")
-			case "pool_tasks_completed_total_" + channel:
+			case "pool_tasks_completed_total":
 				metricValues["tasks_completed"] = metric.GetCounter().GetValue()
 				assert.Equal(t, channel, metric.GetLabel()[0].GetValue(), "Unexpected channel label for completed tasks")
 			}

internal/metrics/mocks.go

@@ -1,7 +1,7 @@
 package metrics
 
 import (
-	"github.com/alitto/pond"
+	"github.com/alitto/pond/v2"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/mock"
 )
@@ -16,7 +16,7 @@ func NewMockMetricsService() *MockMetricsService {
 	return &MockMetricsService{}
 }
 
-func (m *MockMetricsService) RegisterPoolMetrics(channel string, pool *pond.WorkerPool) {
+func (m *MockMetricsService) RegisterPoolMetrics(channel string, pool pond.Pool) {
 	m.Called(channel, pool)
 }
 

internal/services/ingest.go

@@ -105,6 +105,14 @@ func NewIngestService(
 		return nil, errors.New("getLedgersLimit must be greater than 0")
 	}
 
+	// Create worker pools
+	ledgerIndexerPool := pond.NewPool(0)
+	ingestPool := pond.NewPool(0)
+
+	// Register pools with metrics service
+	metricsService.RegisterPoolMetrics("ledger_indexer", ledgerIndexerPool)
+	metricsService.RegisterPoolMetrics("ingest", ingestPool)
+
 	return &ingestService{
 		models:            models,
 		ledgerCursorName:  ledgerCursorName,
@@ -116,8 +124,8 @@ func NewIngestService(
 		metricsService:    metricsService,
 		networkPassphrase: rpcService.NetworkPassphrase(),
 		getLedgersLimit:   getLedgersLimit,
-		ledgerIndexer:     indexer.NewIndexer(rpcService.NetworkPassphrase(), pond.NewPool(0), metricsService),
-		pool:              pond.NewPool(0),
+		ledgerIndexer:     indexer.NewIndexer(rpcService.NetworkPassphrase(), ledgerIndexerPool, metricsService),
+		pool:              ingestPool,
 	}, nil
 }
 

internal/services/ingest_test.go

@@ -114,6 +114,8 @@ func TestGetLedgerTransactions(t *testing.T) {
 	defer dbConnectionPool.Close()
 
 	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("RegisterPoolMetrics", "ledger_indexer", mock.Anything).Return()
+	mockMetricsService.On("RegisterPoolMetrics", "ingest", mock.Anything).Return()
 	models, err := data.NewModels(dbConnectionPool, mockMetricsService)
 	require.NoError(t, err)
 	mockAppTracker := apptracker.MockAppTracker{}
@@ -218,6 +220,8 @@ func TestIngest_LatestSyncedLedgerBehindRPC(t *testing.T) {
 	}()
 
 	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("RegisterPoolMetrics", "ledger_indexer", mock.Anything).Return()
+	mockMetricsService.On("RegisterPoolMetrics", "ingest", mock.Anything).Return()
 	mockMetricsService.On("ObserveDBQueryDuration", "UpdateLatestLedgerSynced", "ingest_store", mock.AnythingOfType("float64")).Once()
 	mockMetricsService.On("IncDBQuery", "UpdateLatestLedgerSynced", "ingest_store").Once()
 	mockMetricsService.On("ObserveDBQueryDuration", "GetLatestLedgerSynced", "ingest_store", mock.AnythingOfType("float64")).Once()
@@ -317,6 +321,8 @@ func TestIngest_LatestSyncedLedgerAheadOfRPC(t *testing.T) {
 	}()
 
 	mockMetricsService := metrics.NewMockMetricsService()
+	mockMetricsService.On("RegisterPoolMetrics", "ledger_indexer", mock.Anything).Return()
+	mockMetricsService.On("RegisterPoolMetrics", "ingest", mock.Anything).Return()
 	models, err := data.NewModels(dbConnectionPool, mockMetricsService)
 	require.NoError(t, err)
 	mockAppTracker := apptracker.MockAppTracker{}
@@ -495,6 +501,8 @@ func Test_ingestService_getLedgerTransactions(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			mockMetricsService := metrics.NewMockMetricsService()
+			mockMetricsService.On("RegisterPoolMetrics", "ledger_indexer", mock.Anything).Return()
+			mockMetricsService.On("RegisterPoolMetrics", "ingest", mock.Anything).Return()
 			defer mockMetricsService.AssertExpectations(t)
 			models, err := data.NewModels(dbConnectionPool, mockMetricsService)
 			require.NoError(t, err)