Repository: stellar/go
Pull request #5209 (2024-02-14 13:01:11)

services/horizon/internal/ingest: Add metrics for ingestion loaders
<!-- 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

Close https://github.com/stellar/go/issues/5154


We do not have any metrics capturing how long it takes to run the ingestion loaders:

https://github.com/stellar/go/blob/release-horizon-v2.28.0/services/horizon/internal/ingest/group_processors.go#L101-L105

This commit adds prometheus metrics for the duration of each loader so we can determine how much of ingestion is spent in the loaders vs the processors. We also added prometheus metrics capturing how many records were queried and inserted during the execution of each loader.


### Known limitations

[N/A]


sreuland: since 2.28.3 went out last unfortunately on a bit of rush, this needs to target master now
tamirms: @sreuland I think I have addressed all your feedback, could you take another look?

Patch:
diff --git a/services/horizon/internal/db2/history/account_loader.go b/services/horizon/internal/db2/history/account_loader.go
index f3946b04..e7e7e908 100644
--- a/services/horizon/internal/db2/history/account_loader.go
+++ b/services/horizon/internal/db2/history/account_loader.go
@@ -39,6 +39,7 @@ type AccountLoader struct {
 	sealed bool
 	set    set.Set[string]
 	ids    map[string]int64
+	stats  LoaderStats
 }
 
 var errSealed = errors.New("cannot register more entries to loader after calling Exec()")
@@ -49,6 +50,7 @@ func NewAccountLoader() *AccountLoader {
 		sealed: false,
 		set:    set.Set[string]{},
 		ids:    map[string]int64{},
+		stats:  LoaderStats{},
 	}
 }
 
@@ -99,6 +101,14 @@ func (a *AccountLoader) lookupKeys(ctx context.Context, q *Q, addresses []string
 	return nil
 }
 
+// LoaderStats describes the result of executing a history lookup id loader
+type LoaderStats struct {
+	// Total is the number of elements registered to the loader
+	Total int
+	// Inserted is the number of elements inserted into the lookup table
+	Inserted int
+}
+
 // Exec will look up all the history account ids for the addresses registered in the loader.
 // If there are no history account ids for a given set of addresses, Exec will insert rows
 // into the history_accounts table to establish a mapping between address and history account id.
@@ -116,6 +126,7 @@ func (a *AccountLoader) Exec(ctx context.Context, session db.SessionInterface) e
 	if err := a.lookupKeys(ctx, q, addresses); err != nil {
 		return err
 	}
+	a.stats.Total += len(addresses)
 
 	insert := 0
 	for _, address := range addresses {
@@ -149,10 +160,21 @@ func (a *AccountLoader) Exec(ctx context.Context, session db.SessionInterface) e
 	if err != nil {
 		return err
 	}
+	a.stats.Inserted += insert
 
 	return a.lookupKeys(ctx, q, addresses)
 }
 
+// Stats returns the number of addresses registered in the loader and the number of addresses
+// inserted into the history_accounts table.
+func (a *AccountLoader) Stats() LoaderStats {
+	return a.stats
+}
+
+func (a *AccountLoader) Name() string {
+	return "AccountLoader"
+}
+
 type bulkInsertField struct {
 	name    string
 	dbType  string
diff --git a/services/horizon/internal/db2/history/asset_loader.go b/services/horizon/internal/db2/history/asset_loader.go
index b5ee9a83..bdf40fb8 100644
--- a/services/horizon/internal/db2/history/asset_loader.go
+++ b/services/horizon/internal/db2/history/asset_loader.go
@@ -60,6 +60,7 @@ type AssetLoader struct {
 	sealed bool
 	set    set.Set[AssetKey]
 	ids    map[AssetKey]int64
+	stats  LoaderStats
 }
 
 // NewAssetLoader will construct a new AssetLoader instance.
@@ -68,6 +69,7 @@ func NewAssetLoader() *AssetLoader {
 		sealed: false,
 		set:    set.Set[AssetKey]{},
 		ids:    map[AssetKey]int64{},
+		stats:  LoaderStats{},
 	}
 }
 
@@ -145,6 +147,7 @@ func (a *AssetLoader) Exec(ctx context.Context, session db.SessionInterface) err
 	if err := a.lookupKeys(ctx, q, keys); err != nil {
 		return err
 	}
+	a.stats.Total += len(keys)
 
 	assetTypes := make([]string, 0, len(a.set)-len(a.ids))
 	assetCodes := make([]string, 0, len(a.set)-len(a.ids))
@@ -196,10 +199,21 @@ func (a *AssetLoader) Exec(ctx context.Context, session db.SessionInterface) err
 	if err != nil {
 		return err
 	}
+	a.stats.Inserted += insert
 
 	return a.lookupKeys(ctx, q, keys)
 }
 
+// Stats returns the number of assets registered in the loader and the number of assets
+// inserted into the history_assets table.
+func (a *AssetLoader) Stats() LoaderStats {
+	return a.stats
+}
+
+func (a *AssetLoader) Name() string {
+	return "AssetLoader"
+}
+
 // AssetLoaderStub is a stub wrapper around AssetLoader which allows
 // you to manually configure the mapping of assets to history asset ids
 type AssetLoaderStub struct {
diff --git a/services/horizon/internal/db2/history/claimable_balance_loader.go b/services/horizon/internal/db2/history/claimable_balance_loader.go
index dd7dee4e..ef18683c 100644
--- a/services/horizon/internal/db2/history/claimable_balance_loader.go
+++ b/services/horizon/internal/db2/history/claimable_balance_loader.go
@@ -34,6 +34,7 @@ type ClaimableBalanceLoader struct {
 	sealed bool
 	set    set.Set[string]
 	ids    map[string]int64
+	stats  LoaderStats
 }
 
 // NewClaimableBalanceLoader will construct a new ClaimableBalanceLoader instance.
@@ -42,6 +43,7 @@ func NewClaimableBalanceLoader() *ClaimableBalanceLoader {
 		sealed: false,
 		set:    set.Set[string]{},
 		ids:    map[string]int64{},
+		stats:  LoaderStats{},
 	}
 }
 
@@ -109,6 +111,7 @@ func (a *ClaimableBalanceLoader) Exec(ctx context.Context, session db.SessionInt
 	if err := a.lookupKeys(ctx, q, ids); err != nil {
 		return err
 	}
+	a.stats.Total += len(ids)
 
 	insert := 0
 	for _, id := range ids {
@@ -142,6 +145,17 @@ func (a *ClaimableBalanceLoader) Exec(ctx context.Context, session db.SessionInt
 	if err != nil {
 		return err
 	}
+	a.stats.Inserted += insert
 
 	return a.lookupKeys(ctx, q, ids)
 }
+
+// Stats returns the number of claimable balances registered in the loader and the number of claimable balances
+// inserted into the history_claimable_balances table.
+func (a *ClaimableBalanceLoader) Stats() LoaderStats {
+	return a.stats
+}
+
+func (a *ClaimableBalanceLoader) Name() string {
+	return "ClaimableBalanceLoader"
+}
diff --git a/services/horizon/internal/db2/history/liquidity_pool_loader.go b/services/horizon/internal/db2/history/liquidity_pool_loader.go
index cf89ae67..d619fa3b 100644
--- a/services/horizon/internal/db2/history/liquidity_pool_loader.go
+++ b/services/horizon/internal/db2/history/liquidity_pool_loader.go
@@ -34,6 +34,7 @@ type LiquidityPoolLoader struct {
 	sealed bool
 	set    set.Set[string]
 	ids    map[string]int64
+	stats  LoaderStats
 }
 
 // NewLiquidityPoolLoader will construct a new LiquidityPoolLoader instance.
@@ -42,6 +43,7 @@ func NewLiquidityPoolLoader() *LiquidityPoolLoader {
 		sealed: false,
 		set:    set.Set[string]{},
 		ids:    map[string]int64{},
+		stats:  LoaderStats{},
 	}
 }
 
@@ -109,6 +111,7 @@ func (a *LiquidityPoolLoader) Exec(ctx context.Context, session db.SessionInterf
 	if err := a.lookupKeys(ctx, q, ids); err != nil {
 		return err
 	}
+	a.stats.Total += len(ids)
 
 	insert := 0
 	for _, id := range ids {
@@ -142,10 +145,21 @@ func (a *LiquidityPoolLoader) Exec(ctx context.Context, session db.SessionInterf
 	if err != nil {
 		return err
 	}
+	a.stats.Inserted += insert
 
 	return a.lookupKeys(ctx, q, ids)
 }
 
+// Stats returns the number of liquidity pools registered in the loader and the number of liquidity pools
+// inserted into the history_liquidity_pools table.
+func (a *LiquidityPoolLoader) Stats() LoaderStats {
+	return a.stats
+}
+
+func (a *LiquidityPoolLoader) Name() string {
+	return "LiquidityPoolLoader"
+}
+
 // LiquidityPoolLoaderStub is a stub wrapper around LiquidityPoolLoader which allows
 // you to manually configure the mapping of liquidity pools to history liquidity ppol ids
 type LiquidityPoolLoaderStub struct {
diff --git a/services/horizon/internal/ingest/fsm.go b/services/horizon/internal/ingest/fsm.go
index 892868e5..c79e11e2 100644
--- a/services/horizon/internal/ingest/fsm.go
+++ b/services/horizon/internal/ingest/fsm.go
@@ -11,6 +11,7 @@ import (
 	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/ingest/ledgerbackend"
+	"github.com/stellar/go/services/horizon/internal/db2/history"
 	"github.com/stellar/go/support/errors"
 	logpkg "github.com/stellar/go/support/log"
 	"github.com/stellar/go/xdr"
@@ -523,6 +524,8 @@ func (r resumeState) run(s *system) (transition, error) {
 	tradeStatsMap := stats.tradeStats.Map()
 	r.addLedgerStatsMetricFromMap(s, "trades", tradeStatsMap)
 	r.addProcessorDurationsMetricFromMap(s, stats.transactionDurations)
+	r.addLoaderDurationsMetricFromMap(s, stats.transactionDurations)
+	r.addLoaderStatsMetric(s, stats.loaderStats)
 
 	// since a single system instance is shared throughout all states,
 	// this will sweep up increments to history archive counters
@@ -573,6 +576,30 @@ func (r resumeState) addProcessorDurationsMetricFromMap(s *system, m map[string]
 	}
 }
 
+func (r resumeState) addLoaderDurationsMetricFromMap(s *system, m map[string]time.Duration) {
+	for loaderName, value := range m {
+		s.Metrics().LoadersRunDurationSummary.
+			With(prometheus.Labels{"name": loaderName}).Observe(value.Seconds())
+	}
+}
+
+func (r resumeState) addLoaderStatsMetric(s *system, loaderSTats map[string]history.LoaderStats) {
+	for loaderName, stats := range loaderSTats {
+		s.Metrics().LoadersStatsSummary.
+			With(prometheus.Labels{
+				"name": loaderName,
+				"stat": "total_queried",
+			}).
+			Observe(float64(stats.Total))
+		s.Metrics().LoadersStatsSummary.
+			With(prometheus.Labels{
+				"name": loaderName,
+				"stat": "total_inserted",
+			}).
+			Observe(float64(stats.Inserted))
+	}
+}
+
 func addHistoryArchiveStatsMetrics(s *system, stats []historyarchive.ArchiveStats) {
 	for _, historyServerStat := range stats {
 		s.Metrics().HistoryArchiveStatsCounter.
diff --git a/services/horizon/internal/ingest/group_processors.go b/services/horizon/internal/ingest/group_processors.go
index 5af3d024..1ea46b4f 100644
--- a/services/horizon/internal/ingest/group_processors.go
+++ b/services/horizon/internal/ingest/group_processors.go
@@ -6,21 +6,22 @@ import (
 	"time"
 
 	"github.com/stellar/go/ingest"
+	"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"
 	"github.com/stellar/go/xdr"
 )
 
-type processorsRunDurations map[string]time.Duration
+type runDurations map[string]time.Duration
 
-func (d processorsRunDurations) AddRunDuration(name string, startTime time.Time) {
+func (d runDurations) AddRunDuration(name string, startTime time.Time) {
 	d[name] += time.Since(startTime)
 }
 
 type groupChangeProcessors struct {
-	processors []horizonChangeProcessor
-	processorsRunDurations
+	processors             []horizonChangeProcessor
+	processorsRunDurations runDurations
 }
 
 func newGroupChangeProcessors(processors []horizonChangeProcessor) *groupChangeProcessors {
@@ -36,7 +37,7 @@ func (g groupChangeProcessors) ProcessChange(ctx context.Context, change ingest.
 		if err := p.ProcessChange(ctx, change); err != nil {
 			return errors.Wrapf(err, "error in %T.ProcessChange", p)
 		}
-		g.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
 	}
 	return nil
 }
@@ -47,15 +48,17 @@ 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.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
 	}
 	return nil
 }
 
 type groupTransactionProcessors struct {
-	processors  []horizonTransactionProcessor
-	lazyLoaders []horizonLazyLoader
-	processorsRunDurations
+	processors                []horizonTransactionProcessor
+	lazyLoaders               []horizonLazyLoader
+	processorsRunDurations    runDurations
+	loaderRunDurations        runDurations
+	loaderStats               map[string]history.LoaderStats
 	transactionStatsProcessor *processors.StatsLedgerTransactionProcessor
 	tradeProcessor            *processors.TradeProcessor
 }
@@ -78,6 +81,8 @@ func newGroupTransactionProcessors(processors []horizonTransactionProcessor,
 	return &groupTransactionProcessors{
 		processors:                processors,
 		processorsRunDurations:    make(map[string]time.Duration),
+		loaderRunDurations:        make(map[string]time.Duration),
+		loaderStats:               make(map[string]history.LoaderStats),
 		lazyLoaders:               lazyLoaders,
 		transactionStatsProcessor: transactionStatsProcessor,
 		tradeProcessor:            tradeProcessor,
@@ -90,7 +95,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.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
 	}
 	return nil
 }
@@ -99,9 +104,16 @@ func (g groupTransactionProcessors) Flush(ctx context.Context, session db.Sessio
 	// need to trigger all lazy loaders to now resolve their future placeholders
 	// with real db values first
 	for _, loader := range g.lazyLoaders {
+		startTime := time.Now()
 		if err := loader.Exec(ctx, session); err != nil {
 			return errors.Wrapf(err, "error during lazy loader resolution, %T.Exec", loader)
 		}
+		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()
 	}
 
 	// now flush each processor which may call loader.GetNow(), which
@@ -111,13 +123,15 @@ 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.AddRunDuration(fmt.Sprintf("%T", p), startTime)
+		g.processorsRunDurations.AddRunDuration(fmt.Sprintf("%T", p), startTime)
 	}
 	return nil
 }
 
 func (g *groupTransactionProcessors) ResetStats() {
 	g.processorsRunDurations = make(map[string]time.Duration)
+	g.loaderRunDurations = make(map[string]time.Duration)
+	g.loaderStats = make(map[string]history.LoaderStats)
 	if g.tradeProcessor != nil {
 		g.tradeProcessor.ResetStats()
 	}
@@ -128,14 +142,14 @@ func (g *groupTransactionProcessors) ResetStats() {
 
 type groupTransactionFilterers struct {
 	filterers []processors.LedgerTransactionFilterer
-	processorsRunDurations
+	runDurations
 	droppedTransactions int64
 }
 
 func newGroupTransactionFilterers(filterers []processors.LedgerTransactionFilterer) *groupTransactionFilterers {
 	return &groupTransactionFilterers{
-		filterers:              filterers,
-		processorsRunDurations: make(map[string]time.Duration),
+		filterers:    filterers,
+		runDurations: make(map[string]time.Duration),
 	}
 }
 
@@ -158,5 +172,5 @@ func (g *groupTransactionFilterers) FilterTransaction(ctx context.Context, tx in
 
 func (g *groupTransactionFilterers) ResetStats() {
 	g.droppedTransactions = 0
-	g.processorsRunDurations = make(map[string]time.Duration)
+	g.runDurations = make(map[string]time.Duration)
 }
diff --git a/services/horizon/internal/ingest/main.go b/services/horizon/internal/ingest/main.go
index ae049800..127d8f02 100644
--- a/services/horizon/internal/ingest/main.go
+++ b/services/horizon/internal/ingest/main.go
@@ -166,6 +166,12 @@ type Metrics struct {
 	// ProcessorsRunDurationSummary exposes processors run durations.
 	ProcessorsRunDurationSummary *prometheus.SummaryVec
 
+	// LoadersRunDurationSummary exposes run durations for the ingestion loaders.
+	LoadersRunDurationSummary *prometheus.SummaryVec
+
+	// LoadersRunDurationSummary exposes stats for the ingestion loaders.
+	LoadersStatsSummary *prometheus.SummaryVec
+
 	// ArchiveRequestCounter counts how many http requests are sent to history server
 	HistoryArchiveStatsCounter *prometheus.CounterVec
 }
@@ -407,6 +413,24 @@ func (s *system) initMetrics() {
 		[]string{"name"},
 	)
 
+	s.metrics.LoadersRunDurationSummary = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Namespace: "horizon", Subsystem: "ingest", Name: "loader_run_duration_seconds",
+			Help:       "run durations of ingestion loaders, sliding window = 10m",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"name"},
+	)
+
+	s.metrics.LoadersStatsSummary = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Namespace: "horizon", Subsystem: "ingest", Name: "loader_stats",
+			Help:       "stats from ingestion loaders, sliding window = 10m",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"name", "stat"},
+	)
+
 	s.metrics.HistoryArchiveStatsCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "horizon", Subsystem: "ingest", Name: "history_archive_stats_total",
@@ -444,6 +468,8 @@ func (s *system) RegisterMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(s.metrics.LedgerStatsCounter)
 	registry.MustRegister(s.metrics.ProcessorsRunDuration)
 	registry.MustRegister(s.metrics.ProcessorsRunDurationSummary)
+	registry.MustRegister(s.metrics.LoadersRunDurationSummary)
+	registry.MustRegister(s.metrics.LoadersStatsSummary)
 	registry.MustRegister(s.metrics.StateVerifyLedgerEntriesCount)
 	registry.MustRegister(s.metrics.HistoryArchiveStatsCounter)
 	s.ledgerBackend = ledgerbackend.WithMetrics(s.ledgerBackend, registry, "horizon")
diff --git a/services/horizon/internal/ingest/processor_runner.go b/services/horizon/internal/ingest/processor_runner.go
index 0023ca31..c9730276 100644
--- a/services/horizon/internal/ingest/processor_runner.go
+++ b/services/horizon/internal/ingest/processor_runner.go
@@ -37,6 +37,8 @@ type horizonTransactionProcessor interface {
 
 type horizonLazyLoader interface {
 	Exec(ctx context.Context, session db.SessionInterface) error
+	Name() string
+	Stats() history.LoaderStats
 }
 
 type statsChangeProcessor struct {
@@ -49,9 +51,11 @@ func (statsChangeProcessor) Commit(ctx context.Context) error {
 
 type ledgerStats struct {
 	changeStats          ingest.StatsChangeProcessorResults
-	changeDurations      processorsRunDurations
+	changeDurations      runDurations
 	transactionStats     processors.StatsLedgerTransactionProcessorResults
-	transactionDurations processorsRunDurations
+	transactionDurations runDurations
+	loaderDurations      runDurations
+	loaderStats          map[string]history.LoaderStats
 	tradeStats           processors.TradeStats
 }
 
@@ -68,8 +72,10 @@ type ProcessorRunnerInterface interface {
 	) (ingest.StatsChangeProcessorResults, error)
 	RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 		transactionStats processors.StatsLedgerTransactionProcessorResults,
-		transactionDurations processorsRunDurations,
+		transactionDurations runDurations,
 		tradeStats processors.TradeStats,
+		loaderDurations runDurations,
+		loaderStats map[string]history.LoaderStats,
 		err error,
 	)
 	RunTransactionProcessorsOnLedgers(ledgers []xdr.LedgerCloseMeta) error
@@ -360,8 +366,10 @@ func (s *ProcessorRunner) streamLedger(ledger xdr.LedgerCloseMeta,
 
 func (s *ProcessorRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 	transactionStats processors.StatsLedgerTransactionProcessorResults,
-	transactionDurations processorsRunDurations,
+	transactionDurations runDurations,
 	tradeStats processors.TradeStats,
+	loaderDurations runDurations,
+	loaderStats map[string]history.LoaderStats,
 	err error,
 ) {
 	// ensure capture of the ledger to history regardless of whether it has transactions.
@@ -394,7 +402,9 @@ func (s *ProcessorRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerClos
 	for key, duration := range groupFilteredOutProcessors.processorsRunDurations {
 		transactionDurations[key] = duration
 	}
-	for key, duration := range groupTransactionFilterers.processorsRunDurations {
+	loaderStats = groupTransactionProcessors.loaderStats
+	loaderDurations = groupTransactionProcessors.loaderRunDurations
+	for key, duration := range groupTransactionFilterers.runDurations {
 		transactionDurations[key] = duration
 	}
 
@@ -499,13 +509,15 @@ func (s *ProcessorRunner) RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 		return
 	}
 
-	transactionStats, transactionDurations, tradeStats, err := s.RunTransactionProcessorsOnLedger(ledger)
+	transactionStats, transactionDurations, tradeStats, loaderDurations, loaderStats, err := s.RunTransactionProcessorsOnLedger(ledger)
 
 	stats.changeStats = changeStatsProcessor.GetResults()
 	stats.changeDurations = groupChangeProcessors.processorsRunDurations
 	stats.transactionStats = transactionStats
 	stats.transactionDurations = transactionDurations
 	stats.tradeStats = tradeStats
+	stats.loaderDurations = loaderDurations
+	stats.loaderStats = loaderStats
 
 	return
 }


Test patch:
diff --git a/services/horizon/internal/db2/history/account_loader_test.go b/services/horizon/internal/db2/history/account_loader_test.go
index 54d2c7a1..ed30b43b 100644
--- a/services/horizon/internal/db2/history/account_loader_test.go
+++ b/services/horizon/internal/db2/history/account_loader_test.go
@@ -31,14 +31,20 @@ func TestAccountLoader(t *testing.T) {
 		assert.Equal(t, future, duplicateFuture)
 	}
 
-	assert.NoError(t, loader.Exec(context.Background(), session))
+	err := loader.Exec(context.Background(), session)
+	assert.NoError(t, err)
+	assert.Equal(t, LoaderStats{
+		Total:    100,
+		Inserted: 100,
+	}, loader.Stats())
 	assert.Panics(t, func() {
 		loader.GetFuture(keypair.MustRandom().Address())
 	})
 
 	q := &Q{session}
 	for _, address := range addresses {
-		internalId, err := loader.GetNow(address)
+		var internalId int64
+		internalId, err = loader.GetNow(address)
 		assert.NoError(t, err)
 		var account Account
 		assert.NoError(t, q.AccountByAddress(context.Background(), &account, address))
@@ -46,7 +52,7 @@ func TestAccountLoader(t *testing.T) {
 		assert.Equal(t, account.Address, address)
 	}
 
-	_, err := loader.GetNow("not present")
+	_, err = loader.GetNow("not present")
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `was not found`)
 }
diff --git a/services/horizon/internal/db2/history/asset_loader_test.go b/services/horizon/internal/db2/history/asset_loader_test.go
index d67163d7..f097561e 100644
--- a/services/horizon/internal/db2/history/asset_loader_test.go
+++ b/services/horizon/internal/db2/history/asset_loader_test.go
@@ -76,14 +76,20 @@ func TestAssetLoader(t *testing.T) {
 		assert.Equal(t, future, duplicateFuture)
 	}
 
-	assert.NoError(t, loader.Exec(context.Background(), session))
+	err := loader.Exec(context.Background(), session)
+	assert.NoError(t, err)
+	assert.Equal(t, LoaderStats{
+		Total:    100,
+		Inserted: 100,
+	}, loader.Stats())
 	assert.Panics(t, func() {
 		loader.GetFuture(AssetKey{Type: "invalid"})
 	})
 
 	q := &Q{session}
 	for _, key := range keys {
-		internalID, err := loader.GetNow(key)
+		var internalID int64
+		internalID, err = loader.GetNow(key)
 		assert.NoError(t, err)
 		var assetXDR xdr.Asset
 		if key.Type == "native" {
@@ -91,12 +97,13 @@ func TestAssetLoader(t *testing.T) {
 		} else {
 			assetXDR = xdr.MustNewCreditAsset(key.Code, key.Issuer)
 		}
-		assetID, err := q.GetAssetID(context.Background(), assetXDR)
+		var assetID int64
+		assetID, err = q.GetAssetID(context.Background(), assetXDR)
 		assert.NoError(t, err)
 		assert.Equal(t, assetID, internalID)
 	}
 
-	_, err := loader.GetNow(AssetKey{})
+	_, err = loader.GetNow(AssetKey{})
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `was not found`)
 }
diff --git a/services/horizon/internal/db2/history/claimable_balance_loader_test.go b/services/horizon/internal/db2/history/claimable_balance_loader_test.go
index 4dd73245..aaf91ccd 100644
--- a/services/horizon/internal/db2/history/claimable_balance_loader_test.go
+++ b/services/horizon/internal/db2/history/claimable_balance_loader_test.go
@@ -2,6 +2,7 @@ package history
 
 import (
 	"context"
+	"database/sql/driver"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -39,7 +40,12 @@ func TestClaimableBalanceLoader(t *testing.T) {
 		assert.Equal(t, future, duplicateFuture)
 	}
 
-	assert.NoError(t, loader.Exec(context.Background(), session))
+	err := loader.Exec(context.Background(), session)
+	assert.NoError(t, err)
+	assert.Equal(t, LoaderStats{
+		Total:    100,
+		Inserted: 100,
+	}, loader.Stats())
 	assert.Panics(t, func() {
 		loader.GetFuture("not-present")
 	})
@@ -47,16 +53,18 @@ func TestClaimableBalanceLoader(t *testing.T) {
 	q := &Q{session}
 	for i, id := range ids {
 		future := futures[i]
-		internalID, err := future.Value()
+		var internalID driver.Value
+		internalID, err = future.Value()
 		assert.NoError(t, err)
-		cb, err := q.ClaimableBalanceByID(context.Background(), id)
+		var cb HistoryClaimableBalance
+		cb, err = q.ClaimableBalanceByID(context.Background(), id)
 		assert.NoError(t, err)
 		assert.Equal(t, cb.BalanceID, id)
 		assert.Equal(t, cb.InternalID, internalID)
 	}
 
 	futureCb := &FutureClaimableBalanceID{id: "not-present", loader: loader}
-	_, err := futureCb.Value()
+	_, err = futureCb.Value()
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `was not found`)
 }
diff --git a/services/horizon/internal/db2/history/liquidity_pool_loader_test.go b/services/horizon/internal/db2/history/liquidity_pool_loader_test.go
index 6e5b4add..25ca8082 100644
--- a/services/horizon/internal/db2/history/liquidity_pool_loader_test.go
+++ b/services/horizon/internal/db2/history/liquidity_pool_loader_test.go
@@ -34,22 +34,29 @@ func TestLiquidityPoolLoader(t *testing.T) {
 		assert.Equal(t, future, duplicateFuture)
 	}
 
-	assert.NoError(t, loader.Exec(context.Background(), session))
+	err := loader.Exec(context.Background(), session)
+	assert.NoError(t, err)
+	assert.Equal(t, LoaderStats{
+		Total:    100,
+		Inserted: 100,
+	}, loader.Stats())
 	assert.Panics(t, func() {
 		loader.GetFuture("not-present")
 	})
 
 	q := &Q{session}
 	for _, id := range ids {
-		internalID, err := loader.GetNow(id)
+		var internalID int64
+		internalID, err = loader.GetNow(id)
 		assert.NoError(t, err)
-		lp, err := q.LiquidityPoolByID(context.Background(), id)
+		var lp HistoryLiquidityPool
+		lp, err = q.LiquidityPoolByID(context.Background(), id)
 		assert.NoError(t, err)
 		assert.Equal(t, lp.PoolID, id)
 		assert.Equal(t, lp.InternalID, internalID)
 	}
 
-	_, err := loader.GetNow("not present")
+	_, err = loader.GetNow("not present")
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), `was not found`)
 }
diff --git a/services/horizon/internal/ingest/main_test.go b/services/horizon/internal/ingest/main_test.go
index 80b5a40e..71c81192 100644
--- a/services/horizon/internal/ingest/main_test.go
+++ b/services/horizon/internal/ingest/main_test.go
@@ -531,14 +531,18 @@ func (m *mockProcessorsRunner) RunAllProcessorsOnLedger(ledger xdr.LedgerCloseMe
 
 func (m *mockProcessorsRunner) RunTransactionProcessorsOnLedger(ledger xdr.LedgerCloseMeta) (
 	processors.StatsLedgerTransactionProcessorResults,
-	processorsRunDurations,
+	runDurations,
 	processors.TradeStats,
+	runDurations,
+	map[string]history.LoaderStats,
 	error,
 ) {
 	args := m.Called(ledger)
 	return args.Get(0).(processors.StatsLedgerTransactionProcessorResults),
-		args.Get(1).(processorsRunDurations),
+		args.Get(1).(runDurations),
 		args.Get(2).(processors.TradeStats),
+		args.Get(3).(runDurations),
+		args.Get(4).(map[string]history.LoaderStats),
 		args.Error(3)
 }