forked from stellar/go
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
services/horizon: Add ingestion processors run duration metrics (stel…
…lar#3224) Add ingestion processors run duration metrics to better understand performance bottlenecks. The new metrics are exposed in /metrics and summarize time spent in each ingestion processor.
- Loading branch information
Showing
12 changed files
with
292 additions
and
113 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,52 +1,83 @@ | ||
| package ingest | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/stellar/go/ingest/io" | ||
| "github.com/stellar/go/support/errors" | ||
| ) | ||
|
|
||
| type horizonChangeProcessor interface { | ||
| io.ChangeProcessor | ||
| // TODO maybe rename to Flush() | ||
| Commit() error | ||
| type processorsRunDurations map[string]time.Duration | ||
|
|
||
| func (d processorsRunDurations) AddRunDuration(name string, startTime time.Time) { | ||
| d[name] += time.Since(startTime) | ||
| } | ||
|
|
||
| type groupChangeProcessors []horizonChangeProcessor | ||
| type groupChangeProcessors struct { | ||
| processors []horizonChangeProcessor | ||
| processorsRunDurations | ||
| } | ||
|
|
||
| func newGroupChangeProcessors(processors []horizonChangeProcessor) *groupChangeProcessors { | ||
| return &groupChangeProcessors{ | ||
| processors: processors, | ||
| processorsRunDurations: make(map[string]time.Duration), | ||
| } | ||
| } | ||
|
|
||
| func (g groupChangeProcessors) ProcessChange(change io.Change) error { | ||
| for _, p := range g { | ||
| for _, p := range g.processors { | ||
| startTime := time.Now() | ||
| if err := p.ProcessChange(change); err != nil { | ||
| return errors.Wrapf(err, "error in %T.ProcessChange", p) | ||
| } | ||
| g.AddRunDuration(fmt.Sprintf("%T", p), startTime) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (g groupChangeProcessors) Commit() error { | ||
| for _, p := range g { | ||
| for _, p := range g.processors { | ||
| startTime := time.Now() | ||
| if err := p.Commit(); err != nil { | ||
| return errors.Wrapf(err, "error in %T.Commit", p) | ||
| } | ||
| g.AddRunDuration(fmt.Sprintf("%T", p), startTime) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| type groupTransactionProcessors []horizonTransactionProcessor | ||
| type groupTransactionProcessors struct { | ||
| processors []horizonTransactionProcessor | ||
| processorsRunDurations | ||
| } | ||
|
|
||
| func newGroupTransactionProcessors(processors []horizonTransactionProcessor) *groupTransactionProcessors { | ||
| return &groupTransactionProcessors{ | ||
| processors: processors, | ||
| processorsRunDurations: make(map[string]time.Duration), | ||
| } | ||
| } | ||
|
|
||
| func (g groupTransactionProcessors) ProcessTransaction(tx io.LedgerTransaction) error { | ||
| for _, p := range g { | ||
| for _, p := range g.processors { | ||
| startTime := time.Now() | ||
| if err := p.ProcessTransaction(tx); err != nil { | ||
| return errors.Wrapf(err, "error in %T.ProcessTransaction", p) | ||
| } | ||
| g.AddRunDuration(fmt.Sprintf("%T", p), startTime) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (g groupTransactionProcessors) Commit() error { | ||
| for _, p := range g { | ||
| for _, p := range g.processors { | ||
| startTime := time.Now() | ||
| if err := p.Commit(); err != nil { | ||
| return errors.Wrapf(err, "error in %T.Commit", p) | ||
| } | ||
| g.AddRunDuration(fmt.Sprintf("%T", p), startTime) | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.