cerc-io/go-ethereum
13
0
Fork 0
Code Issues 9 Pull Requests Packages Projects Releases 66 Wiki Activity

add reporting loop for db metrics

Browse Source
This commit is contained in:
Roy Crihfield 2020-11-20 19:42:46 +08:00
parent 300dae68e5
commit da93bdc0a3
3 changed files with 49 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
statediff/indexer/indexer.go
View File

@ -41,13 +41,17 @@ import (
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
)
var indexerMetrics = RegisterIndexerMetrics(metrics.DefaultRegistry)
var (
indexerMetrics = RegisterIndexerMetrics(metrics.DefaultRegistry)
dbMetrics = RegisterDBMetrics(metrics.DefaultRegistry)
)
// Indexer interface to allow substitution of mocks for testing
type Indexer interface {
PushBlock(block *types.Block, receipts types.Receipts, totalDifficulty *big.Int) (*BlockTx, error)
PushStateNode(tx *BlockTx, stateNode sdtypes.StateNode) error
PushCodeAndCodeHash(tx *BlockTx, codeAndCodeHash sdtypes.CodeAndCodeHash) error
ReportDBMetrics(delay time.Duration, quit <-chan bool)
}
// StateDiffIndexer satisfies the Indexer interface for ethereum statediff objects
@ -72,6 +76,25 @@ type BlockTx struct {
Close func() error
}
// Reporting function to run as goroutine
func (sdi *StateDiffIndexer) ReportDBMetrics(delay time.Duration, quit <-chan bool) {
if !metrics.Enabled {
return
}
ticker := time.NewTicker(delay)
go func() {
for {
select {
case <-ticker.C:
dbMetrics.Update(sdi.dbWriter.db.Stats())
case <-quit:
ticker.Stop()
return
}
}
}()
}
// Pushes and indexes block data in database, except state & storage nodes (includes header, uncles, transactions & receipts)
// Returns an initiated DB transaction which must be Closed via defer to commit or rollback
func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receipts, totalDifficulty *big.Int) (*BlockTx, error) {

44
statediff/indexer/metrics.go
View File

@ -70,7 +70,7 @@ func RegisterIndexerMetrics(reg metrics.Registry) indexerMetricsHandles {
return ctx
}
type dbMetrics struct {
type dbMetricsHandles struct {
// Maximum number of open connections to the database
maxOpen metrics.Gauge
// The number of established connections both in use and idle
@ -82,43 +82,43 @@ type dbMetrics struct {
// The total number of connections waited for
waitedFor metrics.Counter
// The total time blocked waiting for a new connection
blockedSeconds metrics.Counter
blockedMilliseconds metrics.Counter
// The total number of connections closed due to SetMaxIdleConns
closedMaxIdle metrics.Counter
// The total number of connections closed due to SetConnMaxLifetime
closedMaxLifetime metrics.Counter
}
func RegisterDBMetrics(reg metrics.Registry) dbMetrics {
ctx := dbMetrics{
maxOpen: metrics.NewGauge(),
open: metrics.NewGauge(),
inUse: metrics.NewGauge(),
idle: metrics.NewGauge(),
waitedFor: metrics.NewCounter(),
blockedSeconds: metrics.NewCounter(),
closedMaxIdle: metrics.NewCounter(),
closedMaxLifetime: metrics.NewCounter(),
func RegisterDBMetrics(reg metrics.Registry) dbMetricsHandles {
ctx := dbMetricsHandles{
maxOpen: metrics.NewGauge(),
open: metrics.NewGauge(),
inUse: metrics.NewGauge(),
idle: metrics.NewGauge(),
waitedFor: metrics.NewCounter(),
blockedMilliseconds: metrics.NewCounter(),
closedMaxIdle: metrics.NewCounter(),
closedMaxLifetime: metrics.NewCounter(),
}
subsys := "connections"
reg.Register(metricName(subsys, "max_open_desc"), ctx.maxOpen)
reg.Register(metricName(subsys, "open_desc"), ctx.open)
reg.Register(metricName(subsys, "in_use_desc"), ctx.inUse)
reg.Register(metricName(subsys, "idle_desc"), ctx.idle)
reg.Register(metricName(subsys, "waited_for_desc"), ctx.waitedFor)
reg.Register(metricName(subsys, "blocked_seconds_desc"), ctx.blockedSeconds)
reg.Register(metricName(subsys, "closed_max_idle_desc"), ctx.closedMaxIdle)
reg.Register(metricName(subsys, "closed_max_lifetime_desc"), ctx.closedMaxLifetime)
reg.Register(metricName(subsys, "max_open"), ctx.maxOpen)
reg.Register(metricName(subsys, "open"), ctx.open)
reg.Register(metricName(subsys, "in_use"), ctx.inUse)
reg.Register(metricName(subsys, "idle"), ctx.idle)
reg.Register(metricName(subsys, "waited_for"), ctx.waitedFor)
reg.Register(metricName(subsys, "blocked_milliseconds"), ctx.blockedMilliseconds)
reg.Register(metricName(subsys, "closed_max_idle"), ctx.closedMaxIdle)
reg.Register(metricName(subsys, "closed_max_lifetime"), ctx.closedMaxLifetime)
return ctx
}
func (met *dbMetrics) Update(stats sql.DBStats) {
func (met *dbMetricsHandles) Update(stats sql.DBStats) {
met.maxOpen.Update(int64(stats.MaxOpenConnections))
met.open.Update(int64(stats.OpenConnections))
met.inUse.Update(int64(stats.InUse))
met.idle.Update(int64(stats.Idle))
met.waitedFor.Inc(int64(stats.WaitCount))
met.blockedSeconds.Inc(int64(stats.WaitDuration.Seconds()))
met.blockedMilliseconds.Inc(int64(stats.WaitDuration.Milliseconds()))
met.closedMaxIdle.Inc(int64(stats.MaxIdleClosed))
met.closedMaxLifetime.Inc(int64(stats.MaxLifetimeClosed))
}

4
statediff/service.go
View File

@ -22,6 +22,7 @@ import (
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
@ -73,7 +74,7 @@ type IService interface {
// Main event loop for processing state diffs
Loop(chainEventCh chan core.ChainEvent)
// Method to subscribe to receive state diff processing output
Subscribe(id rpc.ID, sub chan<- Payload, quitChanogr chan<- bool, params Params)
Subscribe(id rpc.ID, sub chan<- Payload, quitChan chan<- bool, params Params)
// Method to unsubscribe from state diff processing
Unsubscribe(id rpc.ID) error
// Method to get state diff object at specific block
@ -414,6 +415,7 @@ func (sds *Service) Start() error {
if sds.enableWriteLoop {
log.Info("Starting statediff DB write loop", "params", writeLoopParams)
go sds.WriteLoop(make(chan core.ChainEvent, chainEventChanSize))
go sds.indexer.ReportDBMetrics(5*time.Second, sds.QuitChan)
}
return nil