[FEATURE] [internal] adds metric collection for state change operations

internal/data/statechanges_test.go

@@ -130,8 +130,10 @@ func TestStateChangeModel_BatchInsert(t *testing.T) {
 
 			mockMetricsService := metrics.NewMockMetricsService()
 			mockMetricsService.
-				On("ObserveDBQueryDuration", "BatchInsert", "state_changes", mock.Anything).Return().Once().
-				On("ObserveDBBatchSize", "BatchInsert", "state_changes", mock.Anything).Return().Once().
+				On("ObserveDBQueryDuration", "BatchInsert", "state_changes", mock.Anything).Return().Once()
+			mockMetricsService.
+				On("ObserveDBBatchSize", "BatchInsert", "state_changes", mock.Anything).Return().Once()
+			mockMetricsService.
 				On("IncDBQuery", "BatchInsert", "state_changes").Return().Once()
 
 			m := &StateChangeModel{

internal/indexer/indexer.go

@@ -65,18 +65,20 @@ type Indexer struct {
 	tokenTransferProcessor TokenTransferProcessorInterface
 	processors             []OperationProcessorInterface
 	pool                   pond.Pool
+	metricsService         processors.MetricsServiceInterface
 }
 
-func NewIndexer(networkPassphrase string, ledgerEntryProvider processors.LedgerEntryProvider, pool pond.Pool) *Indexer {
+func NewIndexer(networkPassphrase string, ledgerEntryProvider processors.LedgerEntryProvider, pool pond.Pool, metricsService processors.MetricsServiceInterface) *Indexer {
 	return &Indexer{
 		participantsProcessor:  processors.NewParticipantsProcessor(networkPassphrase),
-		tokenTransferProcessor: processors.NewTokenTransferProcessor(networkPassphrase),
+		tokenTransferProcessor: processors.NewTokenTransferProcessor(networkPassphrase, metricsService),
 		processors: []OperationProcessorInterface{
-			processors.NewEffectsProcessor(networkPassphrase, ledgerEntryProvider),
-			processors.NewContractDeployProcessor(networkPassphrase),
-			contract_processors.NewSACEventsProcessor(networkPassphrase),
+			processors.NewEffectsProcessor(networkPassphrase, ledgerEntryProvider, metricsService),
+			processors.NewContractDeployProcessor(networkPassphrase, metricsService),
+			contract_processors.NewSACEventsProcessor(networkPassphrase, metricsService),
 		},
-		pool: pool,
+		pool:           pool,
+		metricsService: metricsService,
 	}
 }
 

internal/indexer/indexer_test.go

@@ -146,7 +146,7 @@ func TestIndexer_NewIndexer(t *testing.T) {
 	mockLedgerEntryProvider := &MockLedgerEntryProvider{}
 	pool := pond.NewPool(runtime.NumCPU())
 
-	indexer := NewIndexer(networkPassphrase, mockLedgerEntryProvider, pool)
+	indexer := NewIndexer(networkPassphrase, mockLedgerEntryProvider, pool, nil)
 
 	require.NotNil(t, indexer)
 	assert.NotNil(t, indexer.participantsProcessor)

internal/indexer/processors/contract_deploy.go

@@ -3,6 +3,7 @@ package processors
 import (
 	"context"
 	"fmt"
+	"time"
 
 	operation_processor "github.com/stellar/go/processors/operation"
 	"github.com/stellar/go/xdr"
@@ -13,10 +14,14 @@ import (
 // ContractDeployProcessor emits state changes for contract deployments.
 type ContractDeployProcessor struct {
 	networkPassphrase string
+	metricsService    MetricsServiceInterface
 }
 
-func NewContractDeployProcessor(networkPassphrase string) *ContractDeployProcessor {
-	return &ContractDeployProcessor{networkPassphrase: networkPassphrase}
+func NewContractDeployProcessor(networkPassphrase string, metricsService MetricsServiceInterface) *ContractDeployProcessor {
+	return &ContractDeployProcessor{
+		networkPassphrase: networkPassphrase,
+		metricsService:    metricsService,
+	}
 }
 
 func (p *ContractDeployProcessor) Name() string {
@@ -25,13 +30,21 @@ func (p *ContractDeployProcessor) Name() string {
 
 // ProcessOperation emits a state change for each contract deployment (including subinvocations).
 func (p *ContractDeployProcessor) ProcessOperation(_ context.Context, op *operation_processor.TransactionOperationWrapper) ([]types.StateChange, error) {
+	startTime := time.Now()
+	defer func() {
+		if p.metricsService != nil {
+			duration := time.Since(startTime).Seconds()
+			p.metricsService.ObserveStateChangeProcessingDuration("ContractDeployProcessor", duration)
+		}
+	}()
+
 	if op.OperationType() != xdr.OperationTypeInvokeHostFunction {
 		return nil, ErrInvalidOpType
 	}
 	invokeHostOp := op.Operation.Body.MustInvokeHostFunctionOp()
 
 	opID := op.ID()
-	builder := NewStateChangeBuilder(op.Transaction.Ledger.LedgerSequence(), op.LedgerClosed.Unix(), op.Transaction.Hash.HexString(), op.TransactionID()).
+	builder := NewStateChangeBuilder(op.Transaction.Ledger.LedgerSequence(), op.LedgerClosed.Unix(), op.Transaction.Hash.HexString(), op.TransactionID(), p.metricsService).
 		WithOperationID(opID).
 		WithCategory(types.StateChangeCategoryAccount).
 		WithReason(types.StateChangeReasonCreate)

internal/indexer/processors/contract_deploy_test.go

@@ -18,7 +18,7 @@ import (
 
 func Test_ContractDeployProcessor_Process_invalidOpType(t *testing.T) {
 	ctx := context.Background()
-	proc := NewContractDeployProcessor(network.TestNetworkPassphrase)
+	proc := NewContractDeployProcessor(network.TestNetworkPassphrase, nil)
 
 	op := &operation_processor.TransactionOperationWrapper{
 		Operation: xdr.Operation{Body: xdr.OperationBody{Type: xdr.OperationTypePayment}},
@@ -37,7 +37,7 @@ func Test_ContractDeployProcessor_Process_createContract(t *testing.T) {
 
 	ctx := context.Background()
 
-	builder := NewStateChangeBuilder(12345, closeTime.Unix(), "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20", 53021371269120).
+	builder := NewStateChangeBuilder(12345, closeTime.Unix(), "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20", 53021371269120, nil).
 		WithOperationID(53021371269121).
 		WithReason(types.StateChangeReasonCreate).
 		WithCategory(types.StateChangeCategoryAccount)
@@ -136,7 +136,7 @@ func Test_ContractDeployProcessor_Process_createContract(t *testing.T) {
 		}
 	}
 
-	proc := NewContractDeployProcessor(network.TestNetworkPassphrase)
+	proc := NewContractDeployProcessor(network.TestNetworkPassphrase, nil)
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			stateChanges, err := proc.ProcessOperation(ctx, tc.op)
@@ -188,7 +188,7 @@ func Test_ContractDeployProcessor_Process_invokeContract(t *testing.T) {
 		return op
 	}
 
-	builder := NewStateChangeBuilder(12345, closeTime.Unix(), "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20", 53021371269120).
+	builder := NewStateChangeBuilder(12345, closeTime.Unix(), "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20", 53021371269120, nil).
 		WithOperationID(53021371269121).
 		WithReason(types.StateChangeReasonCreate).
 		WithCategory(types.StateChangeCategoryAccount)
@@ -257,7 +257,7 @@ func Test_ContractDeployProcessor_Process_invokeContract(t *testing.T) {
 		}
 	}
 
-	proc := NewContractDeployProcessor(network.TestNetworkPassphrase)
+	proc := NewContractDeployProcessor(network.TestNetworkPassphrase, nil)
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			stateChanges, err := proc.ProcessOperation(ctx, tc.op)

internal/indexer/processors/contracts/sac.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	operation_processor "github.com/stellar/go/processors/operation"
 	"github.com/stellar/go/strkey"
@@ -34,11 +35,13 @@ const (
 
 type SACEventsProcessor struct {
 	networkPassphrase string
+	metricsService    processors.MetricsServiceInterface
 }
 
-func NewSACEventsProcessor(networkPassphrase string) *SACEventsProcessor {
+func NewSACEventsProcessor(networkPassphrase string, metricsService processors.MetricsServiceInterface) *SACEventsProcessor {
 	return &SACEventsProcessor{
 		networkPassphrase: networkPassphrase,
+		metricsService:    metricsService,
 	}
 }
 
@@ -48,6 +51,14 @@ func (p *SACEventsProcessor) Name() string {
 
 // ProcessOperation processes contract events and converts them into state changes.
 func (p *SACEventsProcessor) ProcessOperation(_ context.Context, opWrapper *operation_processor.TransactionOperationWrapper) ([]types.StateChange, error) {
+	startTime := time.Now()
+	defer func() {
+		if p.metricsService != nil {
+			duration := time.Since(startTime).Seconds()
+			p.metricsService.ObserveStateChangeProcessingDuration("SACEventsProcessor", duration)
+		}
+	}()
+
 	if opWrapper.OperationType() != xdr.OperationTypeInvokeHostFunction {
 		return nil, processors.ErrInvalidOpType
 	}
@@ -70,7 +81,7 @@ func (p *SACEventsProcessor) ProcessOperation(_ context.Context, opWrapper *oper
 	}
 
 	stateChanges := make([]types.StateChange, 0)
-	builder := processors.NewStateChangeBuilder(ledgerNumber, ledgerCloseTime, txHash, txID).WithOperationID(opWrapper.ID())
+	builder := processors.NewStateChangeBuilder(ledgerNumber, ledgerCloseTime, txHash, txID, p.metricsService).WithOperationID(opWrapper.ID())
 	for _, event := range contractEvents {
 		// Validate basic contract contractEvent structure
 		if event.Type != xdr.ContractEventTypeContract || event.ContractId == nil || event.Body.V != 0 {

internal/indexer/processors/contracts/sac_test.go

@@ -15,7 +15,7 @@ import (
 )
 
 func TestSACEventsProcessor_ProcessOperation(t *testing.T) {
-	processor := NewSACEventsProcessor(networkPassphrase)
+	processor := NewSACEventsProcessor(networkPassphrase, nil)
 
 	t.Run("V3 Format - set_authorized = true (was unauthorized)", func(t *testing.T) {
 		admin := keypair.MustRandom().Address()

internal/indexer/processors/effects.go

@@ -7,6 +7,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/stellar/go/ingest"
 	effects "github.com/stellar/go/processors/effects"
@@ -63,13 +64,15 @@ var (
 type EffectsProcessor struct {
 	networkPassphrase   string
 	ledgerEntryProvider LedgerEntryProvider // Provider for ledger entry data to avoid import cycles
+	metricsService      MetricsServiceInterface
 }
 
 // NewEffectsProcessor creates a new effects processor for the specified Stellar network.
-func NewEffectsProcessor(networkPassphrase string, ledgerEntryProvider LedgerEntryProvider) *EffectsProcessor {
+func NewEffectsProcessor(networkPassphrase string, ledgerEntryProvider LedgerEntryProvider, metricsService MetricsServiceInterface) *EffectsProcessor {
 	return &EffectsProcessor{
 		networkPassphrase:   networkPassphrase,
 		ledgerEntryProvider: ledgerEntryProvider,
+		metricsService:      metricsService,
 	}
 }
 
@@ -82,6 +85,14 @@ func (p *EffectsProcessor) Name() string {
 // home domain updates, data entry changes, and sponsorship relationship modifications.
 // Returns a slice of state changes representing various account state changes.
 func (p *EffectsProcessor) ProcessOperation(_ context.Context, opWrapper *operation_processor.TransactionOperationWrapper) ([]types.StateChange, error) {
+	startTime := time.Now()
+	defer func() {
+		if p.metricsService != nil {
+			duration := time.Since(startTime).Seconds()
+			p.metricsService.ObserveStateChangeProcessingDuration("EffectsProcessor", duration)
+		}
+	}()
+
 	ledgerCloseTime := opWrapper.Transaction.Ledger.LedgerCloseTime()
 	ledgerNumber := opWrapper.Transaction.Ledger.LedgerSequence()
 	txHash := opWrapper.Transaction.Result.TransactionHash.HexString()
@@ -100,7 +111,7 @@ func (p *EffectsProcessor) ProcessOperation(_ context.Context, opWrapper *operat
 	}
 
 	var stateChanges []types.StateChange
-	masterBuilder := NewStateChangeBuilder(ledgerNumber, ledgerCloseTime, txHash, txID).WithOperationID(opWrapper.ID())
+	masterBuilder := NewStateChangeBuilder(ledgerNumber, ledgerCloseTime, txHash, txID, p.metricsService).WithOperationID(opWrapper.ID())
 	// Process each effect and convert to our internal state change representation
 	for _, effect := range effectOutputs {
 		changeBuilder := masterBuilder.Clone().WithAccount(effect.Address)

internal/indexer/processors/effects_test.go

@@ -50,7 +50,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -115,7 +115,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 		transaction := createTx(setTrustlineFlagsOp, nil, nil, false)
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -159,7 +159,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -199,7 +199,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -244,7 +244,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -283,7 +283,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 		)
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -360,7 +360,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 				},
 			},
 		}, nil)
-		processor := NewEffectsProcessor(networkPassphrase, &mockProvider)
+		processor := NewEffectsProcessor(networkPassphrase, &mockProvider, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -411,7 +411,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 		)
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -454,7 +454,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 		)
 		op, found := transaction.GetOperation(0)
 		require.True(t, found)
-		processor := NewEffectsProcessor(networkPassphrase, nil)
+		processor := NewEffectsProcessor(networkPassphrase, nil, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,
@@ -501,7 +501,7 @@ func TestEffects_ProcessTransaction(t *testing.T) {
 			fmt.Errorf("mock RPC error for testing error handling"),
 		)
 
-		processor := NewEffectsProcessor(networkPassphrase, &mockProvider)
+		processor := NewEffectsProcessor(networkPassphrase, &mockProvider, nil)
 		opWrapper := &operation_processor.TransactionOperationWrapper{
 			Index:          0,
 			Operation:      op,

internal/indexer/processors/metrics.go

@@ -0,0 +1,6 @@
+package processors
+
+// MetricsServiceInterface defines the metrics operations needed by processors
+type MetricsServiceInterface interface {
+	ObserveStateChangeProcessingDuration(processor string, duration float64)
+}

internal/indexer/processors/state_change_builder.go

@@ -13,11 +13,12 @@ import (
 
 // StateChangeBuilder provides a fluent interface for creating state changes
 type StateChangeBuilder struct {
-	base types.StateChange
+	base           types.StateChange
+	metricsService MetricsServiceInterface
 }
 
 // NewStateChangeBuilder creates a new builder with base state change fields
-func NewStateChangeBuilder(ledgerNumber uint32, ledgerCloseTime int64, txHash string, txID int64) *StateChangeBuilder {
+func NewStateChangeBuilder(ledgerNumber uint32, ledgerCloseTime int64, txHash string, txID int64, metricsService MetricsServiceInterface) *StateChangeBuilder {
 	return &StateChangeBuilder{
 		base: types.StateChange{
 			LedgerNumber:    ledgerNumber,
@@ -26,6 +27,7 @@ func NewStateChangeBuilder(ledgerNumber uint32, ledgerCloseTime int64, txHash st
 			TxHash:          txHash,
 			TxID:            txID,
 		},
+		metricsService: metricsService,
 	}
 }
 
@@ -128,6 +130,7 @@ func (b *StateChangeBuilder) Build() types.StateChange {
 		b.base.ToID = b.base.TxID
 	}
 	b.base.SortKey = b.generateSortKey()
+
 	return b.base
 }
 
@@ -178,6 +181,7 @@ func (b *StateChangeBuilder) generateSortKey() string {
 
 func (b *StateChangeBuilder) Clone() *StateChangeBuilder {
 	return &StateChangeBuilder{
-		base: b.base,
+		base:           b.base,
+		metricsService: b.metricsService,
 	}
 }