Repository: stellar/go
Pull request #5216 (2024-02-21 15:46:54)

services/horizon/internal/ingest: Fix transaction processor metrics
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What

When horizon v2.27.0 was deployed to production the ingestion duration of the transaction processor appeared to decrease dramatically:

![image](https://github.com/stellar/go/assets/1445829/43ab068c-bc96-436c-81da-065ce95592ef)

Unfortunately, this is not due to any performance improvements but rather a bug in our reporting of metrics to prometheus.

In the 2.27.0 release ingestion filtering was [enabled by default](https://github.com/stellar/go/commit/f7065378eb39c33cfab69d28095d0f7ae3f68c9d).  When filtering is enabled, Horizon will run an [additional transaction processor](https://github.com/stellar/go/blob/horizon-v2.27.0/services/horizon/internal/ingest/processor_runner.go#L169-L172) on the `history_transactions_filtered_tmp` table. The [transaction processor](https://github.com/stellar/go/blob/horizon-v2.27.0/services/horizon/internal/ingest/processors/transactions_processor.go#L17-L23) for the `history_transactions_filtered_tmp` table has the same type is the regular transaction processor. The type of a transaction is used as the [label](https://github.com/stellar/go/blob/horizon-v2.27.0/services/horizon/internal/ingest/group_processors.go#L71) for the prometheus metric capturing ingestion run duration. So, the metric for the `history_transactions_filtered_tmp` processor was getting reported instead of the normal transactions processor because they both had the [same label](https://github.com/stellar/go/blob/horizon-v2.27.0/services/horizon/internal/ingest/processor_runner.go#L364-L370). 

This PR aims to fix this bug and to prevent bugs like this from occurring again by ensuring that all ingestion processor metrics have unique label names.
 
### Known limitations

[N/A]



Patch:
diff --git a/services/horizon/internal/ingest/filters/account.go b/services/horizon/internal/ingest/filters/account.go
index b2372ae4..08def6a2 100644
--- a/services/horizon/internal/ingest/filters/account.go
+++ b/services/horizon/internal/ingest/filters/account.go
@@ -26,6 +26,10 @@ func NewAccountFilter() AccountFilter {
 	}
 }
 
+func (filter *accountFilter) Name() string {
+	return "filters.accountFilter"
+}
+
 func (filter *accountFilter) RefreshAccountFilter(filterConfig *history.AccountFilterConfig) error {
 	// only need to re-initialize the filter config state(rules) if its cached version(in  memory)
 	// is older than the incoming config version based on lastModified epoch timestamp
diff --git a/services/horizon/internal/ingest/filters/asset.go b/services/horizon/internal/ingest/filters/asset.go
index ee7a44de..0a3a4ca6 100644
--- a/services/horizon/internal/ingest/filters/asset.go
+++ b/services/horizon/internal/ingest/filters/asset.go
@@ -34,6 +34,10 @@ func NewAssetFilter() AssetFilter {
 	}
 }
 
+func (filter *assetFilter) Name() string {
+	return "filters.assetFilter"
+}
+
 func (filter *assetFilter) RefreshAssetFilter(filterConfig *history.AssetFilterConfig) error {
 	// only need to re-initialize the filter config state(rules) if it's cached version(in  memory)
 	// is older than the incoming config version based on lastModified epoch timestamp
diff --git a/services/horizon/internal/ingest/fsm.go b/services/horizon/internal/ingest/fsm.go
index c79e11e2..9c66f30f 100644
--- a/services/horizon/internal/ingest/fsm.go
+++ b/services/horizon/internal/ingest/fsm.go
@@ -567,8 +567,6 @@ func (r resumeState) addLedgerStatsMetricFromMap(s *system, prefix string, m map
 
 func (r resumeState) addProcessorDurationsMetricFromMap(s *system, m map[string]time.Duration) {
 	for processorName, value := range m {
-		// * is not accepted in Prometheus labels
-		processorName = strings.Replace(processorName, "*", "", -1)
 		s.Metrics().ProcessorsRunDuration.
 			With(prometheus.Labels{"name": processorName}).Add(value.Seconds())
 		s.Metrics().ProcessorsRunDurationSummary.
diff --git a/services/horizon/internal/ingest/group_processors.go b/services/horizon/internal/ingest/group_processors.go
index 1ea46b4f..fd0843cc 100644
--- a/services/horizon/internal/ingest/group_processors.go
+++ b/services/horizon/internal/ingest/group_processors.go
@@ -2,7 +2,6 @@ package ingest
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/stellar/go/ingest"
@@ -31,13 +30,17 @@ func newGroupChangeProcessors(processors []horizonChangeProcessor) *groupChangeP
 	}
 }
 
+func (g groupChangeProcessors) Name() string {
+	return "groupChangeProcessors"
+}
+
 func (g groupChangeProcessors) ProcessChange(ctx context.Context, change ingest.Change) error {
 	for _, p := range g.processors {
 		startTime := time.Now()
 		if err := p.ProcessChange(ctx, change); err != nil {
 			return errors.Wrapf(err, "error in %T.ProcessChange", p)
 		}
-		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(p.Name(), startTime)
 	}
 	return nil
 }
@@ -48,7 +51,7 @@ func (g groupChangeProcessors) Commit(ctx context.Context) error {
 		if err := p.Commit(ctx); err != nil {
 			return errors.Wrapf(err, "error in %T.Commit", p)
 		}
-		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(p.Name(), startTime)
 	}
 	return nil
 }
@@ -95,7 +98,7 @@ func (g groupTransactionProcessors) ProcessTransaction(lcm xdr.LedgerCloseMeta,
 		if err := p.ProcessTransaction(lcm, tx); err != nil {
 			return errors.Wrapf(err, "error in %T.ProcessTransaction", p)
 		}
-		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(p.Name(), startTime)
 	}
 	return nil
 }
@@ -110,9 +113,6 @@ func (g groupTransactionProcessors) Flush(ctx context.Context, session db.Sessio
 		}
 		name := loader.Name()
 		g.loaderRunDurations.AddRunDuration(name, startTime)
-		if _, ok := g.loaderStats[name]; ok {
-			return fmt.Errorf("%s is present multiple times", name)
-		}
 		g.loaderStats[name] = loader.Stats()
 	}
 
@@ -123,7 +123,7 @@ func (g groupTransactionProcessors) Flush(ctx context.Context, session db.Sessio
 		if err := p.Flush(ctx, session); err != nil {
 			return errors.Wrapf(err, "error in %T.Flush", p)
 		}
-		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(p.Name(), startTime)
 	}
 	return nil
 }
@@ -153,6 +153,10 @@ func newGroupTransactionFilterers(filterers []processors.LedgerTransactionFilter
 	}
 }
 
+func (g *groupTransactionFilterers) Name() string {
+	return "groupTransactionFilterers"
+}
+
 func (g *groupTransactionFilterers) FilterTransaction(ctx context.Context, tx ingest.LedgerTransaction) (bool, error) {
 	for _, f := range g.filterers {
 		startTime := time.Now()
@@ -160,7 +164,7 @@ func (g *groupTransactionFilterers) FilterTransaction(ctx context.Context, tx in
 		if err != nil {
 			return false, errors.Wrapf(err, "error in %T.FilterTransaction", f)
 		}
-		g.AddRunDuration(fmt.Sprintf("%T", f), startTime)
+		g.AddRunDuration(f.Name(), startTime)
 		if !include {
 			// filter out, we can return early
 			g.droppedTransactions++
diff --git a/services/horizon/internal/ingest/processor_runner.go b/services/horizon/internal/ingest/processor_runner.go
index c9730276..649b345b 100644
--- a/services/horizon/internal/ingest/processor_runner.go
+++ b/services/horizon/internal/ingest/processor_runner.go
@@ -28,10 +28,12 @@ const (
 
 type horizonChangeProcessor interface {
 	processors.ChangeProcessor
+	Name() string
 	Commit(context.Context) error
 }
 
 type horizonTransactionProcessor interface {
+	Name() string
 	processors.LedgerTransactionProcessor
 }
 
@@ -45,6 +47,10 @@ type statsChangeProcessor struct {
 	*ingest.StatsChangeProcessor
 }
 
+func (statsChangeProcessor) Name() string {
+	return "ingest.statsChangeProcessor"
+}
+
 func (statsChangeProcessor) Commit(ctx context.Context) error {
 	return nil
 }
@@ -70,14 +76,6 @@ type ProcessorRunnerInterface interface {
 		ledgerProtocolVersion uint32,
 		bucketListHash xdr.Hash,
 	) (ingest.StatsChangeProcessorResults, error)
-	RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
-		transactionStats processors.StatsLedgerTransactionProcessorResults,
-		transactionDurations runDurations,
-		tradeStats processors.TradeStats,
-		loaderDurations runDurations,
-		loaderStats map[string]history.LoaderStats,
-		err error,
-	)
 	RunTransactionProcessorsOnLedgers(ledgers []xdr.LedgerCloseMeta) error
 	RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 		stats ledgerStats,
@@ -243,6 +241,13 @@ func (s *ProcessorRunner) RunHistoryArchiveIngestion(
 		s.config.NetworkPassphrase,
 	)
 
+	if err := registerChangeProcessors(
+		nameRegistry{},
+		changeProcessor,
+	); err != nil {
+		return ingest.StatsChangeProcessorResults{}, err
+	}
+
 	if checkpointLedger == 1 {
 		if err := changeProcessor.ProcessChange(s.ctx, ingest.GenesisChange(s.config.NetworkPassphrase)); err != nil {
 			return changeStats.GetResults(), errors.Wrap(err, "Error ingesting genesis ledger")
@@ -364,7 +369,7 @@ func (s *ProcessorRunner) streamLedger(ledger xdr.LedgerCloseMeta,
 	return nil
 }
 
-func (s *ProcessorRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
+func (s *ProcessorRunner) runTransactionProcessorsOnLedger(registry nameRegistry, ledger xdr.LedgerCloseMeta) (
 	transactionStats processors.StatsLedgerTransactionProcessorResults,
 	transactionDurations runDurations,
 	tradeStats processors.TradeStats,
@@ -380,6 +385,15 @@ func (s *ProcessorRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerClos
 	groupFilteredOutProcessors := s.buildFilteredOutProcessor()
 	groupTransactionProcessors := s.buildTransactionProcessor(ledgersProcessor)
 
+	if err = registerTransactionProcessors(
+		registry,
+		groupTransactionFilterers,
+		groupFilteredOutProcessors,
+		groupTransactionProcessors,
+	); err != nil {
+		return
+	}
+
 	err = s.streamLedger(ledger,
 		groupTransactionFilterers,
 		groupFilteredOutProcessors,
@@ -413,6 +427,64 @@ func (s *ProcessorRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerClos
 	return
 }
 
+// nameRegistry ensures all ingestion components have a unique name
+// for metrics reporting
+type nameRegistry map[string]struct{}
+
+func (n nameRegistry) add(name string) error {
+	if _, ok := n[name]; ok {
+		return fmt.Errorf("%s is duplicated", name)
+	}
+	n[name] = struct{}{}
+	return nil
+}
+
+func registerChangeProcessors(
+	registry nameRegistry,
+	group *groupChangeProcessors,
+) error {
+	for _, p := range group.processors {
+		if err := registry.add(p.Name()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func registerTransactionProcessors(
+	registry nameRegistry,
+	groupTransactionFilterers *groupTransactionFilterers,
+	groupFilteredOutProcessors *groupTransactionProcessors,
+	groupTransactionProcessors *groupTransactionProcessors,
+) error {
+	for _, f := range groupTransactionFilterers.filterers {
+		if err := registry.add(f.Name()); err != nil {
+			return err
+		}
+	}
+	for _, p := range groupTransactionProcessors.processors {
+		if err := registry.add(p.Name()); err != nil {
+			return err
+		}
+	}
+	for _, l := range groupTransactionProcessors.lazyLoaders {
+		if err := registry.add(l.Name()); err != nil {
+			return err
+		}
+	}
+	for _, p := range groupFilteredOutProcessors.processors {
+		if err := registry.add(p.Name()); err != nil {
+			return err
+		}
+	}
+	for _, l := range groupFilteredOutProcessors.lazyLoaders {
+		if err := registry.add(l.Name()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (s *ProcessorRunner) RunTransactionProcessorsOnLedgers(ledgers []xdr.LedgerCloseMeta) (err error) {
 	ledgersProcessor := processors.NewLedgerProcessor(s.historyQ.NewLedgerBatchInsertBuilder(), CurrentVersion)
 
@@ -504,12 +576,21 @@ func (s *ProcessorRunner) RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 		ledger.LedgerSequence(),
 		s.config.NetworkPassphrase,
 	)
+
+	registry := nameRegistry{}
+	if err = registerChangeProcessors(
+		registry,
+		groupChangeProcessors,
+	); err != nil {
+		return
+	}
+
 	err = s.runChangeProcessorOnLedger(groupChangeProcessors, ledger)
 	if err != nil {
 		return
 	}
 
-	transactionStats, transactionDurations, tradeStats, loaderDurations, loaderStats, err := s.RunTransactionProcessorsOnLedger(ledger)
+	transactionStats, transactionDurations, tradeStats, loaderDurations, loaderStats, err := s.runTransactionProcessorsOnLedger(registry, ledger)
 
 	stats.changeStats = changeStatsProcessor.GetResults()
 	stats.changeDurations = groupChangeProcessors.processorsRunDurations
diff --git a/services/horizon/internal/ingest/processors/account_data_processor.go b/services/horizon/internal/ingest/processors/account_data_processor.go
index f774700f..50679356 100644
--- a/services/horizon/internal/ingest/processors/account_data_processor.go
+++ b/services/horizon/internal/ingest/processors/account_data_processor.go
@@ -27,6 +27,10 @@ func (p *AccountDataProcessor) reset() {
 	p.batchInsertBuilder = p.dataQ.NewAccountDataBatchInsertBuilder()
 }
 
+func (p *AccountDataProcessor) Name() string {
+	return "processors.AccountDataProcessor"
+}
+
 func (p *AccountDataProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
 	// We're interested in data only
 	if change.Type != xdr.LedgerEntryTypeData {
diff --git a/services/horizon/internal/ingest/processors/accounts_processor.go b/services/horizon/internal/ingest/processors/accounts_processor.go
index 681b7d38..17128f80 100644
--- a/services/horizon/internal/ingest/processors/accounts_processor.go
+++ b/services/horizon/internal/ingest/processors/accounts_processor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/guregu/null/zero"
+
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
 	"github.com/stellar/go/support/errors"
@@ -28,6 +29,10 @@ func (p *AccountsProcessor) reset() {
 	p.batchInsertBuilder = p.accountsQ.NewAccountsBatchInsertBuilder()
 }
 
+func (p *AccountsProcessor) Name() string {
+	return "processors.AccountsProcessor"
+}
+
 func (p *AccountsProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
 	if change.Type != xdr.LedgerEntryTypeAccount {
 		return nil
diff --git a/services/horizon/internal/ingest/processors/asset_stats_processor.go b/services/horizon/internal/ingest/processors/asset_stats_processor.go
index c3da42b7..9423f245 100644
--- a/services/horizon/internal/ingest/processors/asset_stats_processor.go
+++ b/services/horizon/internal/ingest/processors/asset_stats_processor.go
@@ -44,6 +44,10 @@ func NewAssetStatsProcessor(
 	return p
 }
 
+func (p *AssetStatsProcessor) Name() string {
+	return "processors.AssetStatsProcessor"
+}
+
 func (p *AssetStatsProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
 	if change.Type != xdr.LedgerEntryTypeLiquidityPool &&
 		change.Type != xdr.LedgerEntryTypeClaimableBalance &&
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
index 5d985e69..de729f96 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
@@ -27,6 +27,10 @@ func NewClaimableBalancesChangeProcessor(Q history.QClaimableBalances) *Claimabl
 	return p
 }
 
+func (p *ClaimableBalancesChangeProcessor) Name() string {
+	return "processors.ClaimableBalancesChangeProcessor"
+}
+
 func (p *ClaimableBalancesChangeProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
 	p.claimantsInsertBuilder = p.qClaimableBalances.NewClaimableBalanceClaimantBatchInsertBuilder()
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_transaction_processor.go b/services/horizon/internal/ingest/processors/claimable_balances_transaction_processor.go
index 394d2e0f..80261f93 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_transaction_processor.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_transaction_processor.go
@@ -29,6 +29,10 @@ func NewClaimableBalancesTransactionProcessor(
 	}
 }
 
+func (p *ClaimableBalancesTransactionProcessor) Name() string {
+	return "processors.ClaimableBalancesTransactionProcessor"
+}
+
 func (p *ClaimableBalancesTransactionProcessor) ProcessTransaction(
 	lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction,
 ) error {
diff --git a/services/horizon/internal/ingest/processors/effects_processor.go b/services/horizon/internal/ingest/processors/effects_processor.go
index 34e9f916..e6e0e4e4 100644
--- a/services/horizon/internal/ingest/processors/effects_processor.go
+++ b/services/horizon/internal/ingest/processors/effects_processor.go
@@ -42,6 +42,10 @@ func NewEffectProcessor(
 	}
 }
 
+func (p *EffectProcessor) Name() string {
+	return "processors.EffectProcessor"
+}
+
 func (p *EffectProcessor) ProcessTransaction(
 	lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction,
 ) error {
diff --git a/services/horizon/internal/ingest/processors/ledgers_processor.go b/services/horizon/internal/ingest/processors/ledgers_processor.go
index 5ecfb2ad..0ed2d78b 100644
--- a/services/horizon/internal/ingest/processors/ledgers_processor.go
+++ b/services/horizon/internal/ingest/processors/ledgers_processor.go
@@ -32,6 +32,10 @@ func NewLedgerProcessor(batch history.LedgerBatchInsertBuilder, ingestVersion in
 	}
 }
 
+func (p *LedgersProcessor) Name() string {
+	return "processors.LedgersProcessor"
+}
+
 func (p *LedgersProcessor) ProcessLedger(lcm xdr.LedgerCloseMeta) *ledgerInfo {
 	sequence := lcm.LedgerSequence()
 	entry, ok := p.ledgers[sequence]
diff --git a/services/horizon/internal/ingest/processors/liquidity_pools_change_processor.go b/services/horizon/internal/ingest/processors/liquidity_pools_change_processor.go
index c5e52522..3dce45aa 100644
--- a/services/horizon/internal/ingest/processors/liquidity_pools_change_processor.go
+++ b/services/horizon/internal/ingest/processors/liquidity_pools_change_processor.go
@@ -24,6 +24,10 @@ func NewLiquidityPoolsChangeProcessor(Q history.QLiquidityPools, sequence uint32
 	return p
 }
 
+func (p *LiquidityPoolsChangeProcessor) Name() string {
+	return "processors.LiquidityPoolsChangeProcessor"
+}
+
 func (p *LiquidityPoolsChangeProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
 }
diff --git a/services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor.go b/services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor.go
index c721f9e4..f902904f 100644
--- a/services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor.go
+++ b/services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor.go
@@ -30,6 +30,10 @@ func NewLiquidityPoolsTransactionProcessor(
 	}
 }
 
+func (p *LiquidityPoolsTransactionProcessor) Name() string {
+	return "processors.LiquidityPoolsTransactionProcessor"
+}
+
 func (p *LiquidityPoolsTransactionProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) error {
 	err := p.addTransactionLiquidityPools(lcm.LedgerSequence(), transaction)
 	if err != nil {
diff --git a/services/horizon/internal/ingest/processors/main.go b/services/horizon/internal/ingest/processors/main.go
index 94f83f3f..2b6c1cc8 100644
--- a/services/horizon/internal/ingest/processors/main.go
+++ b/services/horizon/internal/ingest/processors/main.go
@@ -5,6 +5,7 @@ import (
 	"io"
 
 	"github.com/guregu/null"
+
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/support/db"
 	"github.com/stellar/go/support/errors"
@@ -26,6 +27,7 @@ type LedgerTransactionProcessor interface {
 }
 
 type LedgerTransactionFilterer interface {
+	Name() string
 	FilterTransaction(ctx context.Context, transaction ingest.LedgerTransaction) (bool, error)
 }
 
diff --git a/services/horizon/internal/ingest/processors/offers_processor.go b/services/horizon/internal/ingest/processors/offers_processor.go
index 13ee130d..2a3e529b 100644
--- a/services/horizon/internal/ingest/processors/offers_processor.go
+++ b/services/horizon/internal/ingest/processors/offers_processor.go
@@ -27,6 +27,10 @@ func NewOffersProcessor(offersQ history.QOffers, sequence uint32) *OffersProcess
 	return p
 }
 
+func (p *OffersProcessor) Name() string {
+	return "processors.OffersProcessor"
+}
+
 func (p *OffersProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
 	p.insertBatchBuilder = p.offersQ.NewOffersBatchInsertBuilder()
diff --git a/services/horizon/internal/ingest/processors/operations_processor.go b/services/horizon/internal/ingest/processors/operations_processor.go
index 8ad02314..252d5a6b 100644
--- a/services/horizon/internal/ingest/processors/operations_processor.go
+++ b/services/horizon/internal/ingest/processors/operations_processor.go
@@ -33,6 +33,10 @@ func NewOperationProcessor(batch history.OperationBatchInsertBuilder, network st
 	}
 }
 
+func (p *OperationProcessor) Name() string {
+	return "processors.OperationProcessor"
+}
+
 // ProcessTransaction process the given transaction
 func (p *OperationProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) error {
 	for i, op := range transaction.Envelope.Operations() {
diff --git a/services/horizon/internal/ingest/processors/participants_processor.go b/services/horizon/internal/ingest/processors/participants_processor.go
index 7d4ae7fe..e889e747 100644
--- a/services/horizon/internal/ingest/processors/participants_processor.go
+++ b/services/horizon/internal/ingest/processors/participants_processor.go
@@ -99,6 +99,10 @@ func participantsForMeta(
 	return participants, nil
 }
 
+func (p *ParticipantsProcessor) Name() string {
+	return "processors.ParticipantsProcessor"
+}
+
 func (p *ParticipantsProcessor) addTransactionParticipants(
 	sequence uint32,
 	transaction ingest.LedgerTransaction,
diff --git a/services/horizon/internal/ingest/processors/signers_processor.go b/services/horizon/internal/ingest/processors/signers_processor.go
index b7286752..b2118382 100644
--- a/services/horizon/internal/ingest/processors/signers_processor.go
+++ b/services/horizon/internal/ingest/processors/signers_processor.go
@@ -30,6 +30,10 @@ func NewSignersProcessor(
 	return p
 }
 
+func (p *SignersProcessor) Name() string {
+	return "processors.SignersProcessor"
+}
+
 func (p *SignersProcessor) reset() {
 	p.batchInsertBuilder = p.signersQ.NewAccountSignersBatchInsertBuilder()
 	p.cache = ingest.NewChangeCompactor()
diff --git a/services/horizon/internal/ingest/processors/stats_ledger_transaction_processor.go b/services/horizon/internal/ingest/processors/stats_ledger_transaction_processor.go
index 83188e02..9efe51dc 100644
--- a/services/horizon/internal/ingest/processors/stats_ledger_transaction_processor.go
+++ b/services/horizon/internal/ingest/processors/stats_ledger_transaction_processor.go
@@ -141,6 +141,10 @@ func (p *StatsLedgerTransactionProcessor) ProcessTransaction(lcm xdr.LedgerClose
 	return nil
 }
 
+func (p *StatsLedgerTransactionProcessor) Name() string {
+	return "processors.StatsLedgerTransactionProcessor"
+}
+
 func (p *StatsLedgerTransactionProcessor) GetResults() StatsLedgerTransactionProcessorResults {
 	return p.results
 }
diff --git a/services/horizon/internal/ingest/processors/trades_processor.go b/services/horizon/internal/ingest/processors/trades_processor.go
index 66c6aeb8..131531fc 100644
--- a/services/horizon/internal/ingest/processors/trades_processor.go
+++ b/services/horizon/internal/ingest/processors/trades_processor.go
@@ -55,6 +55,10 @@ func (stats *TradeStats) Map() map[string]interface{} {
 	}
 }
 
+func (p *TradeProcessor) Name() string {
+	return "processors.TradeProcessor"
+}
+
 // ProcessTransaction process the given transaction
 func (p *TradeProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) (err error) {
 	if !transaction.Result.Successful() {
diff --git a/services/horizon/internal/ingest/processors/transactions_processor.go b/services/horizon/internal/ingest/processors/transactions_processor.go
index 28aae6f3..da615674 100644
--- a/services/horizon/internal/ingest/processors/transactions_processor.go
+++ b/services/horizon/internal/ingest/processors/transactions_processor.go
@@ -13,12 +13,14 @@ import (
 type TransactionProcessor struct {
 	batch      history.TransactionBatchInsertBuilder
 	skipTxmeta bool
+	name       string
 }
 
 func NewTransactionFilteredTmpProcessor(batch history.TransactionBatchInsertBuilder, skipTxmeta bool) *TransactionProcessor {
 	return &TransactionProcessor{
 		batch:      batch,
 		skipTxmeta: skipTxmeta,
+		name:       "processors.TransactionFilteredTmpProcessor",
 	}
 }
 
@@ -26,9 +28,14 @@ func NewTransactionProcessor(batch history.TransactionBatchInsertBuilder, skipTx
 	return &TransactionProcessor{
 		batch:      batch,
 		skipTxmeta: skipTxmeta,
+		name:       "processors.TransactionProcessor",
 	}
 }
 
+func (p *TransactionProcessor) Name() string {
+	return p.name
+}
+
 func (p *TransactionProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) error {
 	elidedTransaction := transaction
 
diff --git a/services/horizon/internal/ingest/processors/trust_lines_processor.go b/services/horizon/internal/ingest/processors/trust_lines_processor.go
index 3152ef0d..a7cae533 100644
--- a/services/horizon/internal/ingest/processors/trust_lines_processor.go
+++ b/services/horizon/internal/ingest/processors/trust_lines_processor.go
@@ -22,6 +22,10 @@ func NewTrustLinesProcessor(trustLinesQ history.QTrustLines) *TrustLinesProcesso
 	return p
 }
 
+func (p *TrustLinesProcessor) Name() string {
+	return "processors.TrustLinesProcessor"
+}
+
 func (p *TrustLinesProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
 	p.batchInsertBuilder = p.trustLinesQ.NewTrustLinesBatchInsertBuilder()


Test patch:
diff --git a/services/horizon/internal/ingest/group_processors_test.go b/services/horizon/internal/ingest/group_processors_test.go
index 71e50d39..9fc5a3ac 100644
--- a/services/horizon/internal/ingest/group_processors_test.go
+++ b/services/horizon/internal/ingest/group_processors_test.go
@@ -23,6 +23,10 @@ type mockHorizonChangeProcessor struct {
 	mock.Mock
 }
 
+func (m *mockHorizonChangeProcessor) Name() string {
+	return "mockHorizonChangeProcessor"
+}
+
 func (m *mockHorizonChangeProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
 	args := m.Called(ctx, change)
 	return args.Error(0)
@@ -39,6 +43,10 @@ type mockHorizonTransactionProcessor struct {
 	mock.Mock
 }
 
+func (m *mockHorizonTransactionProcessor) Name() string {
+	return "mockHorizonTransactionProcessor"
+}
+
 func (m *mockHorizonTransactionProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, transaction ingest.LedgerTransaction) error {
 	args := m.Called(lcm, transaction)
 	return args.Error(0)
diff --git a/services/horizon/internal/ingest/main_test.go b/services/horizon/internal/ingest/main_test.go
index 71c81192..3ba66ef4 100644
--- a/services/horizon/internal/ingest/main_test.go
+++ b/services/horizon/internal/ingest/main_test.go
@@ -15,7 +15,6 @@ import (
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/ingest/ledgerbackend"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
-	"github.com/stellar/go/services/horizon/internal/ingest/processors"
 	"github.com/stellar/go/support/db"
 	"github.com/stellar/go/support/errors"
 	logpkg "github.com/stellar/go/support/log"
@@ -529,23 +528,6 @@ func (m *mockProcessorsRunner) RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMe
 		args.Error(1)
 }
 
-func (m *mockProcessorsRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
-	processors.StatsLedgerTransactionProcessorResults,
-	runDurations,
-	processors.TradeStats,
-	runDurations,
-	map[string]history.LoaderStats,
-	error,
-) {
-	args := m.Called(ledger)
-	return args.Get(0).(processors.StatsLedgerTransactionProcessorResults),
-		args.Get(1).(runDurations),
-		args.Get(2).(processors.TradeStats),
-		args.Get(3).(runDurations),
-		args.Get(4).(map[string]history.LoaderStats),
-		args.Error(3)
-}
-
 func (m *mockProcessorsRunner) RunTransactionProcessorsOnLedgers(ledgers []xdr.LedgerCloseMeta) error {
 	args := m.Called(ledgers)
 	return args.Error(0)