From 779f2fbaa930c4972392e64c843f8978474417c8 Mon Sep 17 00:00:00 2001 From: Thomas E Lackey Date: Wed, 11 Jan 2023 19:16:21 -0600 Subject: [PATCH 1/5] Add additional statediff metrics and logging. --- statediff/api.go | 15 +++++-- statediff/metrics.go | 25 ++++++++++- statediff/metrics_helpers.go | 87 ++++++++++++++++++++++++++++++++++++ statediff/service.go | 19 ++++---- 4 files changed, 131 insertions(+), 15 deletions(-) create mode 100644 statediff/metrics_helpers.go diff --git a/statediff/api.go b/statediff/api.go index 0a7c5bba8..7dc1947a1 100644 --- a/statediff/api.go +++ b/statediff/api.go @@ -18,7 +18,6 @@ package statediff import ( "context" - "github.com/ethereum/go-ethereum/statediff/types" "github.com/ethereum/go-ethereum/common" @@ -142,12 +141,22 @@ func (api *PublicStateDiffAPI) StreamCodeAndCodeHash(ctx context.Context, blockN // WriteStateDiffAt writes a state diff object directly to DB at the specific blockheight func (api *PublicStateDiffAPI) WriteStateDiffAt(ctx context.Context, blockNumber uint64, params Params) error { - return api.sds.WriteStateDiffAt(blockNumber, params) + var err error + start, logger := countApiRequestBegin("writeStateDiffAt", blockNumber) + defer countApiRequestEnd(start, logger, err) + + err = api.sds.WriteStateDiffAt(blockNumber, params) + return err } // WriteStateDiffFor writes a state diff object directly to DB for the specific block hash func (api *PublicStateDiffAPI) WriteStateDiffFor(ctx context.Context, blockHash common.Hash, params Params) error { - return api.sds.WriteStateDiffFor(blockHash, params) + var err error + start, logger := countApiRequestBegin("writeStateDiffFor", blockHash.Hex()) + defer countApiRequestEnd(start, logger, err) + + err = api.sds.WriteStateDiffFor(blockHash, params) + return err } // WatchAddress changes the list of watched addresses to which the direct indexing is restricted according to given operation diff --git a/statediff/metrics.go b/statediff/metrics.go index e67499b94..e94ee4de3 100644 --- a/statediff/metrics.go +++ b/statediff/metrics.go @@ -17,15 +17,16 @@ package statediff import ( - "strings" - "github.com/ethereum/go-ethereum/metrics" + "strings" ) const ( namespace = "statediff" ) +var defaultStatediffMetrics = RegisterStatediffMetrics(metrics.DefaultRegistry) + // Build a fully qualified metric name func metricName(subsystem, name string) string { if name == "" { @@ -58,6 +59,14 @@ type statediffMetricsHandles struct { knownGapErrorStart metrics.Gauge // A known gaps end block which had an error being written to the DB knownGapErrorEnd metrics.Gauge + + apiRequests metrics.Counter + apiRequestsUnderway metrics.Counter + + failed metrics.Counter + succeeded metrics.Counter + underway metrics.Counter + totalProcessingTime metrics.Gauge } func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles { @@ -71,6 +80,12 @@ func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles { knownGapEnd: metrics.NewGauge(), knownGapErrorStart: metrics.NewGauge(), knownGapErrorEnd: metrics.NewGauge(), + apiRequests: metrics.NewCounter(), + apiRequestsUnderway: metrics.NewCounter(), + failed: metrics.NewCounter(), + succeeded: metrics.NewCounter(), + underway: metrics.NewCounter(), + totalProcessingTime: metrics.NewGauge(), } subsys := "service" reg.Register(metricName(subsys, "last_sync_height"), ctx.lastSyncHeight) @@ -82,5 +97,11 @@ func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles { reg.Register(metricName(subsys, "known_gaps_end"), ctx.knownGapEnd) reg.Register(metricName(subsys, "known_gaps_error_start"), ctx.knownGapErrorStart) reg.Register(metricName(subsys, "known_gaps_error_end"), ctx.knownGapErrorEnd) + reg.Register(metricName(subsys, "api_requests"), ctx.apiRequests) + reg.Register(metricName(subsys, "api_requests_underway"), ctx.apiRequestsUnderway) + reg.Register(metricName(subsys, "failed"), ctx.failed) + reg.Register(metricName(subsys, "succeeded"), ctx.succeeded) + reg.Register(metricName(subsys, "underway"), ctx.underway) + reg.Register(metricName(subsys, "total_processing_time"), ctx.totalProcessingTime) return ctx } diff --git a/statediff/metrics_helpers.go b/statediff/metrics_helpers.go new file mode 100644 index 000000000..f6b9b7909 --- /dev/null +++ b/statediff/metrics_helpers.go @@ -0,0 +1,87 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package statediff + +import ( + "fmt" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" + "time" +) + +func countStateDiffBegin(block *types.Block) (time.Time, log.Logger) { + start := time.Now() + logger := log.New("hash", block.Hash().Hex(), "number", block.NumberU64()) + + defaultStatediffMetrics.underway.Inc(1) + logger.Debug(fmt.Sprintf("writeStateDiff BEGIN [underway=%d, succeeded=%d, failed=%d, time=%d]", + defaultStatediffMetrics.underway.Count(), + defaultStatediffMetrics.succeeded.Count(), + defaultStatediffMetrics.failed.Count(), + defaultStatediffMetrics.totalProcessingTime.Value(), + )) + + return start, logger +} + +func countStateDiffEnd(start time.Time, logger log.Logger, err error) time.Duration { + duration := time.Since(start) + defaultStatediffMetrics.underway.Dec(1) + if nil == err { + defaultStatediffMetrics.succeeded.Inc(1) + } else { + defaultStatediffMetrics.failed.Inc(1) + } + defaultStatediffMetrics.totalProcessingTime.Inc(duration.Milliseconds()) + + logger.Debug(fmt.Sprintf("writeStateDiff END (duration=%dms, err=%t) [underway=%d, succeeded=%d, failed=%d, total_time=%d]", + duration.Milliseconds(), nil != err, + defaultStatediffMetrics.underway.Count(), + defaultStatediffMetrics.succeeded.Count(), + defaultStatediffMetrics.failed.Count(), + defaultStatediffMetrics.totalProcessingTime.Value(), + )) + return duration +} + +func countApiRequestBegin(methodName string, blockHashOrNumber interface{}) (time.Time, log.Logger) { + start := time.Now() + logger := log.New(methodName, blockHashOrNumber) + + defaultStatediffMetrics.apiRequests.Inc(1) + defaultStatediffMetrics.apiRequestsUnderway.Inc(1) + + logger.Debug(fmt.Sprintf("statediff API BEGIN [api_underway=%d, api_total=%d])", + defaultStatediffMetrics.apiRequestsUnderway.Count(), + defaultStatediffMetrics.apiRequests.Count(), + )) + + return start, logger +} + +func countApiRequestEnd(start time.Time, logger log.Logger, err error) time.Duration { + duration := time.Since(start) + defaultStatediffMetrics.apiRequestsUnderway.Dec(1) + + logger.Debug(fmt.Sprintf("statediff API END (duration=%dms, err=%t) [api_underway=%d, api_total=%d]", + duration.Milliseconds(), nil != err, + defaultStatediffMetrics.apiRequestsUnderway.Count(), + defaultStatediffMetrics.apiRequests.Count(), + )) + + return duration +} diff --git a/statediff/service.go b/statediff/service.go index b73247647..947a78e70 100644 --- a/statediff/service.go +++ b/statediff/service.go @@ -36,7 +36,6 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" @@ -71,8 +70,6 @@ var writeLoopParams = ParamsWithMutex{ }, } -var statediffMetrics = RegisterStatediffMetrics(metrics.DefaultRegistry) - type blockChain interface { SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription CurrentBlock() *types.Block @@ -199,7 +196,7 @@ func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params errorState: false, writeFilePath: params.KnownGapsFilePath, db: db, - statediffMetrics: statediffMetrics, + statediffMetrics: defaultStatediffMetrics, sqlFileWaitingForWrite: false, } if indexerConfigAvailable { @@ -294,8 +291,8 @@ func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) { for { select { case chainEvent := <-chainEventCh: - statediffMetrics.lastEventHeight.Update(int64(chainEvent.Block.Number().Uint64())) - statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) + defaultStatediffMetrics.lastEventHeight.Update(int64(chainEvent.Block.Number().Uint64())) + defaultStatediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) chainEventFwd <- chainEvent case err := <-errCh: log.Error("Error from chain event subscription", "error", err) @@ -333,7 +330,7 @@ func (sds *Service) writeGenesisStateDiff(currBlock *types.Block, workerId uint) genesisBlockNumber, "error", err.Error(), "worker", workerId) return } - statediffMetrics.lastStatediffHeight.Update(genesisBlockNumber) + defaultStatediffMetrics.lastStatediffHeight.Update(genesisBlockNumber) } func (sds *Service) writeLoopWorker(params workerParams) { @@ -394,7 +391,7 @@ func (sds *Service) writeLoopWorker(params workerParams) { } // TODO: how to handle with concurrent workers - statediffMetrics.lastStatediffHeight.Update(int64(currentBlock.Number().Uint64())) + defaultStatediffMetrics.lastStatediffHeight.Update(int64(currentBlock.Number().Uint64())) case <-sds.QuitChan: log.Info("Quitting the statediff writing process", "worker", params.id) return @@ -412,7 +409,7 @@ func (sds *Service) Loop(chainEventCh chan core.ChainEvent) { select { //Notify chain event channel of events case chainEvent := <-chainEventCh: - statediffMetrics.serviceLoopChannelLen.Update(int64(len(chainEventCh))) + defaultStatediffMetrics.serviceLoopChannelLen.Update(int64(len(chainEventCh))) log.Debug("Loop(): chain event received", "event", chainEvent) // if we don't have any subscribers, do not process a statediff if atomic.LoadInt32(&sds.subscribers) == 0 { @@ -844,6 +841,7 @@ func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params Params) erro // Writes a state diff from the current block, parent state root, and provided params func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, params Params) error { + start, logger := countStateDiffBegin(block) // log.Info("Writing state diff", "block height", block.Number().Uint64()) var totalDifficulty *big.Int var receipts types.Receipts @@ -862,8 +860,9 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p // defer handling of commit/rollback for any return case defer func() { if err := tx.Submit(err); err != nil { - log.Error("batch transaction submission failed", "err", err) + logger.Error("batch transaction submission failed", "err", err) } + countStateDiffEnd(start, logger, err) }() output := func(node types2.StateNode) error { return sds.indexer.PushStateNode(tx, node, block.Hash().String()) -- 2.45.2 From 571051ef174102c2bb759b3b15c683ea32e44499 Mon Sep 17 00:00:00 2001 From: Thomas E Lackey Date: Wed, 11 Jan 2023 20:12:16 -0600 Subject: [PATCH 2/5] goimports --- statediff/api.go | 1 + statediff/metrics.go | 3 ++- statediff/metrics_helpers.go | 3 ++- 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/statediff/api.go b/statediff/api.go index 7dc1947a1..55b9b1f9a 100644 --- a/statediff/api.go +++ b/statediff/api.go @@ -18,6 +18,7 @@ package statediff import ( "context" + "github.com/ethereum/go-ethereum/statediff/types" "github.com/ethereum/go-ethereum/common" diff --git a/statediff/metrics.go b/statediff/metrics.go index e94ee4de3..f3461ca21 100644 --- a/statediff/metrics.go +++ b/statediff/metrics.go @@ -17,8 +17,9 @@ package statediff import ( - "github.com/ethereum/go-ethereum/metrics" "strings" + + "github.com/ethereum/go-ethereum/metrics" ) const ( diff --git a/statediff/metrics_helpers.go b/statediff/metrics_helpers.go index f6b9b7909..ff44fd8c6 100644 --- a/statediff/metrics_helpers.go +++ b/statediff/metrics_helpers.go @@ -18,9 +18,10 @@ package statediff import ( "fmt" + "time" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/log" - "time" ) func countStateDiffBegin(block *types.Block) (time.Time, log.Logger) { -- 2.45.2 From 7c73a4b405c4c6b0c65e4ff6c8df640f91c1cf35 Mon Sep 17 00:00:00 2001 From: Thomas E Lackey Date: Wed, 11 Jan 2023 23:38:30 -0600 Subject: [PATCH 3/5] minor formatting --- statediff/metrics_helpers.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/statediff/metrics_helpers.go b/statediff/metrics_helpers.go index ff44fd8c6..c83c88009 100644 --- a/statediff/metrics_helpers.go +++ b/statediff/metrics_helpers.go @@ -49,13 +49,14 @@ func countStateDiffEnd(start time.Time, logger log.Logger, err error) time.Durat } defaultStatediffMetrics.totalProcessingTime.Inc(duration.Milliseconds()) - logger.Debug(fmt.Sprintf("writeStateDiff END (duration=%dms, err=%t) [underway=%d, succeeded=%d, failed=%d, total_time=%d]", + logger.Debug(fmt.Sprintf("writeStateDiff END (duration=%dms, err=%t) [underway=%d, succeeded=%d, failed=%d, total_time=%dms]", duration.Milliseconds(), nil != err, defaultStatediffMetrics.underway.Count(), defaultStatediffMetrics.succeeded.Count(), defaultStatediffMetrics.failed.Count(), defaultStatediffMetrics.totalProcessingTime.Value(), )) + return duration } @@ -66,7 +67,7 @@ func countApiRequestBegin(methodName string, blockHashOrNumber interface{}) (tim defaultStatediffMetrics.apiRequests.Inc(1) defaultStatediffMetrics.apiRequestsUnderway.Inc(1) - logger.Debug(fmt.Sprintf("statediff API BEGIN [api_underway=%d, api_total=%d])", + logger.Debug(fmt.Sprintf("statediff API BEGIN [underway=%d, requests=%d])", defaultStatediffMetrics.apiRequestsUnderway.Count(), defaultStatediffMetrics.apiRequests.Count(), )) @@ -78,7 +79,7 @@ func countApiRequestEnd(start time.Time, logger log.Logger, err error) time.Dura duration := time.Since(start) defaultStatediffMetrics.apiRequestsUnderway.Dec(1) - logger.Debug(fmt.Sprintf("statediff API END (duration=%dms, err=%t) [api_underway=%d, api_total=%d]", + logger.Debug(fmt.Sprintf("statediff API END (duration=%dms, err=%t) [underway=%d, requests=%d]", duration.Milliseconds(), nil != err, defaultStatediffMetrics.apiRequestsUnderway.Count(), defaultStatediffMetrics.apiRequests.Count(), -- 2.45.2 From 5bd106bcfa9065de5b90ed8f3f9bbe126521e1b0 Mon Sep 17 00:00:00 2001 From: Thomas E Lackey Date: Wed, 11 Jan 2023 23:44:33 -0600 Subject: [PATCH 4/5] minor formatting --- statediff/metrics_helpers.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/statediff/metrics_helpers.go b/statediff/metrics_helpers.go index c83c88009..2bebfe253 100644 --- a/statediff/metrics_helpers.go +++ b/statediff/metrics_helpers.go @@ -29,7 +29,7 @@ func countStateDiffBegin(block *types.Block) (time.Time, log.Logger) { logger := log.New("hash", block.Hash().Hex(), "number", block.NumberU64()) defaultStatediffMetrics.underway.Inc(1) - logger.Debug(fmt.Sprintf("writeStateDiff BEGIN [underway=%d, succeeded=%d, failed=%d, time=%d]", + logger.Debug(fmt.Sprintf("writeStateDiff BEGIN [underway=%d, succeeded=%d, failed=%d, total_time=%dms]", defaultStatediffMetrics.underway.Count(), defaultStatediffMetrics.succeeded.Count(), defaultStatediffMetrics.failed.Count(), -- 2.45.2 From 6b469b438f8625415b0ed181f7be4710f59338f8 Mon Sep 17 00:00:00 2001 From: Thomas E Lackey Date: Thu, 12 Jan 2023 10:49:12 -0600 Subject: [PATCH 5/5] Merge --- statediff/service.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/statediff/service.go b/statediff/service.go index 0165a1baf..a147ca1c9 100644 --- a/statediff/service.go +++ b/statediff/service.go @@ -267,13 +267,13 @@ func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) { for { select { case chainEvent := <-chainEventCh: - lastHeight := statediffMetrics.lastEventHeight.Value() + lastHeight := defaultStatediffMetrics.lastEventHeight.Value() nextHeight := int64(chainEvent.Block.Number().Uint64()) if nextHeight-lastHeight != 1 { log.Warn("Statediffing service received block out-of-order", "next height", nextHeight, "last height", lastHeight) } - statediffMetrics.lastEventHeight.Update(nextHeight) - statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) + defaultStatediffMetrics.lastEventHeight.Update(nextHeight) + defaultStatediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh))) chainEventFwd <- chainEvent case err := <-errCh: log.Error("Error from chain event subscription", "error", err) -- 2.45.2