[FEATURE] [internal] adds metrics for ingestion phases (#351)

* adds metrics for ingestion phases

* fixes formatting for metrics

* removes current account counters

* adds counter for operations processed

Author: aristidesstaffieri

internal/indexer/indexer.go

@@ -25,6 +25,7 @@ type IndexerBufferInterface interface {
 	GetTransactionsParticipants() map[string]set.Set[string]
 	GetOperationsParticipants() map[int64]set.Set[string]
 	GetNumberOfTransactions() int
+	GetNumberOfOperations() int
 	GetTransactions() []types.Transaction
 	GetOperations() []types.Operation
 	GetStateChanges() []types.StateChange

internal/indexer/indexer_buffer.go

@@ -105,6 +105,15 @@ func (b *IndexerBuffer) GetNumberOfTransactions() int {
 	return len(b.txByHash)
 }
 
+// GetNumberOfOperations returns the count of unique operations in the buffer.
+// Thread-safe: uses read lock.
+func (b *IndexerBuffer) GetNumberOfOperations() int {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	return len(b.opByID)
+}
+
 // GetTransactions returns all unique transactions.
 // Thread-safe: uses read lock.
 func (b *IndexerBuffer) GetTransactions() []types.Transaction {

internal/metrics/metrics.go

@@ -36,6 +36,13 @@ type MetricsService interface {
 	ObserveDBTransactionDuration(status string, duration float64)
 	ObserveDBBatchSize(operation, table string, size int)
 	IncSignatureVerificationExpired(expiredSeconds float64)
+	// Ingestion Phase Metrics
+	ObserveIngestionPhaseDuration(phase string, duration float64)
+	IncIngestionLedgersProcessed(count int)
+	IncIngestionTransactionsProcessed(count int)
+	IncIngestionOperationsProcessed(count int)
+	ObserveIngestionBatchSize(size int)
+	ObserveIngestionParticipantsCount(count int)
 	// GraphQL Metrics
 	ObserveGraphQLFieldDuration(operationName, fieldName string, duration float64)
 	IncGraphQLField(operationName, fieldName string, success bool)
@@ -83,6 +90,14 @@ type metricsService struct {
 	// Signature Verification Metrics
 	signatureVerificationExpired *prometheus.CounterVec
 
+	// Ingestion Phase Metrics
+	ingestionPhaseDuration     *prometheus.SummaryVec
+	ingestionLedgersProcessed  prometheus.Counter
+	ingestionTransactionsTotal prometheus.Counter
+	ingestionOperationsTotal   prometheus.Counter
+	ingestionBatchSize         prometheus.Histogram
+	ingestionParticipantsCount prometheus.Histogram
+
 	// GraphQL Metrics
 	graphqlFieldDuration *prometheus.SummaryVec
 	graphqlFieldsTotal   *prometheus.CounterVec
@@ -261,6 +276,48 @@ func NewMetricsService(db *sqlx.DB) MetricsService {
 		[]string{"expired_seconds"},
 	)
 
+	// Ingestion Phase Metrics
+	m.ingestionPhaseDuration = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "ingestion_phase_duration_seconds",
+			Help:       "Duration of each ingestion phase",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"phase"},
+	)
+	m.ingestionLedgersProcessed = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "ingestion_ledgers_processed_total",
+			Help: "Total number of ledgers processed during ingestion",
+		},
+	)
+	m.ingestionTransactionsTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "ingestion_transactions_processed_total",
+			Help: "Total number of transactions processed during ingestion",
+		},
+	)
+	m.ingestionOperationsTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "ingestion_operations_processed_total",
+			Help: "Total number of operations processed during ingestion",
+		},
+	)
+	m.ingestionBatchSize = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "ingestion_batch_size",
+			Help:    "Number of ledgers processed per ingestion batch",
+			Buckets: prometheus.ExponentialBuckets(1, 2, 8), // 1, 2, 4, 8, 16, 32, 64, 128
+		},
+	)
+	m.ingestionParticipantsCount = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "ingestion_participants_count",
+			Help:    "Number of unique participants per ingestion batch",
+			Buckets: prometheus.ExponentialBuckets(1, 2, 12), // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048
+		},
+	)
+
 	// GraphQL Metrics
 	m.graphqlFieldDuration = prometheus.NewSummaryVec(
 		prometheus.SummaryOpts{
@@ -322,6 +379,12 @@ func (m *metricsService) registerMetrics() {
 		m.dbTxnDuration,
 		m.dbBatchSize,
 		m.signatureVerificationExpired,
+		m.ingestionPhaseDuration,
+		m.ingestionLedgersProcessed,
+		m.ingestionTransactionsTotal,
+		m.ingestionOperationsTotal,
+		m.ingestionBatchSize,
+		m.ingestionParticipantsCount,
 		m.graphqlFieldDuration,
 		m.graphqlFieldsTotal,
 		m.graphqlComplexity,
@@ -514,6 +577,31 @@ func (m *metricsService) IncSignatureVerificationExpired(expiredSeconds float64)
 	m.signatureVerificationExpired.WithLabelValues(fmt.Sprintf("%fs", expiredSeconds)).Inc()
 }
 
+// Ingestion Phase Metrics
+func (m *metricsService) ObserveIngestionPhaseDuration(phase string, duration float64) {
+	m.ingestionPhaseDuration.WithLabelValues(phase).Observe(duration)
+}
+
+func (m *metricsService) IncIngestionLedgersProcessed(count int) {
+	m.ingestionLedgersProcessed.Add(float64(count))
+}
+
+func (m *metricsService) IncIngestionTransactionsProcessed(count int) {
+	m.ingestionTransactionsTotal.Add(float64(count))
+}
+
+func (m *metricsService) IncIngestionOperationsProcessed(count int) {
+	m.ingestionOperationsTotal.Add(float64(count))
+}
+
+func (m *metricsService) ObserveIngestionBatchSize(size int) {
+	m.ingestionBatchSize.Observe(float64(size))
+}
+
+func (m *metricsService) ObserveIngestionParticipantsCount(count int) {
+	m.ingestionParticipantsCount.Observe(float64(count))
+}
+
 // GraphQL Metrics
 func (m *metricsService) ObserveGraphQLFieldDuration(operationName, fieldName string, duration float64) {
 	m.graphqlFieldDuration.WithLabelValues(operationName, fieldName).Observe(duration)

internal/metrics/metrics_test.go

@@ -461,6 +461,153 @@ func TestDBBatchSizeMetrics(t *testing.T) {
 	})
 }
 
+func TestIngestionPhaseMetrics(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+
+	ms := NewMetricsService(db)
+
+	t.Run("ingestion phase duration metrics", func(t *testing.T) {
+		// Record durations for different phases
+		ms.ObserveIngestionPhaseDuration("fetch_ledgers", 0.5)
+		ms.ObserveIngestionPhaseDuration("collect_transaction_data", 1.2)
+		ms.ObserveIngestionPhaseDuration("fetch_existing_accounts", 0.3)
+		ms.ObserveIngestionPhaseDuration("process_and_buffer", 2.1)
+		ms.ObserveIngestionPhaseDuration("merge_buffers", 0.1)
+		ms.ObserveIngestionPhaseDuration("db_insertion", 1.5)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_phase_duration_seconds" {
+				found = true
+				// Should have 6 metrics (one for each phase)
+				assert.Equal(t, 6, len(mf.GetMetric()))
+
+				// Verify each phase is recorded
+				phaseLabels := make(map[string]bool)
+				for _, metric := range mf.GetMetric() {
+					labels := make(map[string]string)
+					for _, label := range metric.GetLabel() {
+						labels[label.GetName()] = label.GetValue()
+					}
+					phaseLabels[labels["phase"]] = true
+					assert.Equal(t, uint64(1), metric.GetSummary().GetSampleCount())
+				}
+
+				assert.True(t, phaseLabels["fetch_ledgers"])
+				assert.True(t, phaseLabels["collect_transaction_data"])
+				assert.True(t, phaseLabels["fetch_existing_accounts"])
+				assert.True(t, phaseLabels["process_and_buffer"])
+				assert.True(t, phaseLabels["merge_buffers"])
+				assert.True(t, phaseLabels["db_insertion"])
+			}
+		}
+		assert.True(t, found)
+	})
+
+	t.Run("ingestion ledgers processed counter", func(t *testing.T) {
+		ms.IncIngestionLedgersProcessed(10)
+		ms.IncIngestionLedgersProcessed(5)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_ledgers_processed_total" {
+				found = true
+				metric := mf.GetMetric()[0]
+				assert.Equal(t, float64(15), metric.GetCounter().GetValue())
+			}
+		}
+		assert.True(t, found)
+	})
+
+	t.Run("ingestion transactions processed counter", func(t *testing.T) {
+		ms.IncIngestionTransactionsProcessed(100)
+		ms.IncIngestionTransactionsProcessed(50)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_transactions_processed_total" {
+				found = true
+				metric := mf.GetMetric()[0]
+				assert.Equal(t, float64(150), metric.GetCounter().GetValue())
+			}
+		}
+		assert.True(t, found)
+	})
+
+	t.Run("ingestion operations processed counter", func(t *testing.T) {
+		ms.IncIngestionOperationsProcessed(200)
+		ms.IncIngestionOperationsProcessed(75)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_operations_processed_total" {
+				found = true
+				metric := mf.GetMetric()[0]
+				assert.Equal(t, float64(275), metric.GetCounter().GetValue())
+			}
+		}
+		assert.True(t, found)
+	})
+
+	t.Run("ingestion batch size histogram", func(t *testing.T) {
+		// Record various batch sizes
+		ms.ObserveIngestionBatchSize(1)
+		ms.ObserveIngestionBatchSize(10)
+		ms.ObserveIngestionBatchSize(50)
+		ms.ObserveIngestionBatchSize(25)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_batch_size" {
+				found = true
+				metric := mf.GetMetric()[0]
+				histogram := metric.GetHistogram()
+				assert.Equal(t, uint64(4), histogram.GetSampleCount())
+				assert.Equal(t, float64(86), histogram.GetSampleSum()) // 1 + 10 + 50 + 25 = 86
+			}
+		}
+		assert.True(t, found)
+	})
+
+	t.Run("ingestion participants count histogram", func(t *testing.T) {
+		// Record various participant counts
+		ms.ObserveIngestionParticipantsCount(5)
+		ms.ObserveIngestionParticipantsCount(100)
+		ms.ObserveIngestionParticipantsCount(500)
+
+		metricFamilies, err := ms.GetRegistry().Gather()
+		require.NoError(t, err)
+
+		found := false
+		for _, mf := range metricFamilies {
+			if mf.GetName() == "ingestion_participants_count" {
+				found = true
+				metric := mf.GetMetric()[0]
+				histogram := metric.GetHistogram()
+				assert.Equal(t, uint64(3), histogram.GetSampleCount())
+				assert.Equal(t, float64(605), histogram.GetSampleSum()) // 5 + 100 + 500 = 605
+			}
+		}
+		assert.True(t, found)
+	})
+}
+
 func TestPoolMetrics(t *testing.T) {
 	db := setupTestDB(t)
 	defer db.Close()

internal/metrics/mocks.go

@@ -113,6 +113,32 @@ func (m *MockMetricsService) IncSignatureVerificationExpired(expiredSeconds floa
 	m.Called(expiredSeconds)
 }
 
+// Ingestion Phase Metrics
+func (m *MockMetricsService) ObserveIngestionPhaseDuration(phase string, duration float64) {
+	m.Called(phase, duration)
+}
+
+func (m *MockMetricsService) IncIngestionLedgersProcessed(count int) {
+	m.Called(count)
+}
+
+func (m *MockMetricsService) IncIngestionTransactionsProcessed(count int) {
+	m.Called(count)
+}
+
+func (m *MockMetricsService) IncIngestionOperationsProcessed(count int) {
+	m.Called(count)
+}
+
+func (m *MockMetricsService) ObserveIngestionBatchSize(size int) {
+	m.Called(size)
+}
+
+func (m *MockMetricsService) ObserveIngestionParticipantsCount(count int) {
+	m.Called(count)
+}
+
+// GraphQL Metrics
 func (m *MockMetricsService) ObserveGraphQLFieldDuration(operationName, fieldName string, duration float64) {
 	m.Called(operationName, fieldName, duration)
 }