diff --git a/cmd/evmconnect.go b/cmd/evmconnect.go index 4c0fa20..460be39 100644 --- a/cmd/evmconnect.go +++ b/cmd/evmconnect.go @@ -110,6 +110,11 @@ func run() error { return err } + // Emit the block listener metrics into the metrics registry of the manager + if err := c.BlockListener().InitMetrics(ctx, m.MetricsRegistry()); err != nil { + return err + } + // Setup signal handling to cancel the context, which shuts down the API Server signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) go func() { diff --git a/internal/msgs/en_error_messages.go b/internal/msgs/en_error_messages.go index 2f0a093..5fc6b8c 100644 --- a/internal/msgs/en_error_messages.go +++ b/internal/msgs/en_error_messages.go @@ -91,4 +91,5 @@ var ( MsgTransactionEstimateTooLargeForBlock = ffe("FF23071", "Gas estimate %s (scaled at %.2f from estimate %s) too large for the current block gas limit %s") MsgMonitoredHeadLengthInvalid = ffe("FF23072", "Monitored head length must be greater than or equal to 1 value=%d") MsgUnknownJSONFormatOptionValue = ffe("FF23073", "Unknown value '%s' for JSON formatting option '%s'. Supported values: %s") + MsgMetricsInitFail = ffe("FF23074", "Failed to initialize metrics for subsystem '%s'") ) diff --git a/mocks/ethblocklistenermocks/block_listener.go b/mocks/ethblocklistenermocks/block_listener.go index 09ce492..cdd8601 100644 --- a/mocks/ethblocklistenermocks/block_listener.go +++ b/mocks/ethblocklistenermocks/block_listener.go @@ -6,6 +6,7 @@ import ( context "context" fftypes "github.com/hyperledger-firefly/common/pkg/fftypes" + metric "github.com/hyperledger-firefly/common/pkg/metric" ethblocklistener "github.com/hyperledger-firefly/evmconnect/pkg/ethblocklistener" ethrpc "github.com/hyperledger-firefly/evmconnect/pkg/ethrpc" ethtypes "github.com/hyperledger-firefly/signer/pkg/ethtypes" @@ -324,6 +325,24 @@ func (_m *BlockListener) GetMonitoredHeadLength() int { return r0 } +// InitMetrics provides a mock function with given fields: ctx, registry +func (_m *BlockListener) InitMetrics(ctx context.Context, registry metric.MetricsRegistry) error { + ret := _m.Called(ctx, registry) + + if len(ret) == 0 { + panic("no return value specified for InitMetrics") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, metric.MetricsRegistry) error); ok { + r0 = rf(ctx, registry) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // ReconcileConfirmationsForTransaction provides a mock function with given fields: ctx, txHash, existingConfirmations, targetConfirmationCount func (_m *BlockListener) ReconcileConfirmationsForTransaction(ctx context.Context, txHash string, existingConfirmations []*ethrpc.MinimalBlockInfo, targetConfirmationCount uint64) (*ethblocklistener.ConfirmationUpdateResult, *ethrpc.TxReceiptJSONRPC, error) { ret := _m.Called(ctx, txHash, existingConfirmations, targetConfirmationCount) diff --git a/pkg/ethblocklistener/blocklistener.go b/pkg/ethblocklistener/blocklistener.go index 93964ea..e571007 100644 --- a/pkg/ethblocklistener/blocklistener.go +++ b/pkg/ethblocklistener/blocklistener.go @@ -27,6 +27,7 @@ import ( "github.com/hyperledger-firefly/common/pkg/fftypes" "github.com/hyperledger-firefly/common/pkg/i18n" "github.com/hyperledger-firefly/common/pkg/log" + "github.com/hyperledger-firefly/common/pkg/metric" "github.com/hyperledger-firefly/common/pkg/retry" "github.com/hyperledger-firefly/common/pkg/wsclient" "github.com/hyperledger-firefly/evmconnect/internal/msgs" @@ -89,6 +90,7 @@ type BlockListener interface { WaitClosed() GetBackend() rpcbackend.RPC UTSetBackend(rpcbackend.RPC) + InitMetrics(ctx context.Context, registry metric.MetricsRegistry) error } func toMinimalBlockInfoList(blocks []*ethrpc.BlockInfoJSONRPC) []*ethrpc.MinimalBlockInfo { @@ -142,6 +144,10 @@ type blockListener struct { // headBlockNumber mode: last head value sent on the block listener channel (only written from listenLoop) currentChainHead uint64 + + // metrics are optional - only emitted once InitMetrics has been called + metricsLock sync.RWMutex + metrics metric.MetricsManager } func NewBlockListener(ctx context.Context, retry *retry.Retry, conf *BlockListenerConfig, httpConf *ffresty.Config, wsConf *wsclient.WSConfig) (bl BlockListener, err error) { @@ -295,26 +301,28 @@ func (bl *blockListener) establishBlockHeightWithRetry() error { } // Now get the block height - var hexBlockHeight ethtypes.HexInteger - rpcErr := bl.backend.CallRPC(bl.ctx, &hexBlockHeight, "eth_blockNumber") - if rpcErr != nil { - log.L(bl.ctx).Warnf("Block height could not be obtained: %s", rpcErr.Message) - return true, rpcErr.Error() + head, err := bl.queryBlockHeightFromRPC() + if err != nil { + log.L(bl.ctx).Warnf("Block height could not be obtained: %s", err) + return true, err } - bl.setHighestBlock(hexBlockHeight.BigInt().Uint64()) + bl.setHighestBlock(head) return false, nil }) } -// refreshHighestBlockFromRPC updates highestBlock from eth_blockNumber. Caller must not hold canonicalChainLock. -func (bl *blockListener) refreshHighestBlockFromRPC() (uint64, error) { +// queryBlockHeightFromRPC queries eth_blockNumber and returns the result, without updating any listener +// state. Caller must not hold canonicalChainLock. The height the node reports is recorded on the target block height gauge. +func (bl *blockListener) queryBlockHeightFromRPC() (uint64, error) { var hexBlockHeight ethtypes.HexInteger rpcErr := bl.backend.CallRPC(bl.ctx, &hexBlockHeight, "eth_blockNumber") if rpcErr != nil { + bl.incPollFailureMetric("eth_blockNumber") return 0, rpcErr.Error() } head := hexBlockHeight.BigInt().Uint64() + bl.setBlockHeightMetric(metricTargetBlockHeight, head) return head, nil } @@ -367,10 +375,17 @@ func (bl *blockListener) listenLoop() { } } + // In full chain tracking mode, the loop below never queries the height the node reports, so we refresh + // it here for the target metric. Done ahead of the filter calls. + if bl.ChainTrackingMode != ffcapi.ChainTrackingModeLight { + bl.refreshTargetBlockHeightMetric() + } + if filter == "" { err := bl.backend.CallRPC(bl.ctx, &filter, "eth_newBlockFilter") if err != nil { log.L(bl.ctx).Errorf("Failed to establish new block filter: %s", err.Message) + bl.incPollFailureMetric("eth_newBlockFilter") failCount++ continue } @@ -393,6 +408,7 @@ func (bl *blockListener) listenLoop() { gapPotential = true } log.L(bl.ctx).Errorf("Failed to query block filter changes: %s", rpcErr.Message) + bl.incPollFailureMetric("eth_getFilterChanges") failCount++ continue } @@ -400,17 +416,20 @@ func (bl *blockListener) listenLoop() { } if bl.ChainTrackingMode == ffcapi.ChainTrackingModeLight { - head, err := bl.refreshHighestBlockFromRPC() + head, err := bl.queryBlockHeightFromRPC() if err != nil { log.L(bl.ctx).Errorf("Failed to refresh chain head: %s", err) failCount++ continue } + // In light mode there is no canonical chain being built, so the head we dispatch to + // consumers is what we report as the canonical height if head == bl.currentChainHead { failCount = 0 continue } bl.currentChainHead = head + bl.setBlockHeightMetric(metricCanonicalBlockHeight, bl.currentChainHead) update := &ffcapi.BlockHashEvent{GapPotential: false, Created: fftypes.Now(), HeadBlockNumber: bl.currentChainHead} bl.consumerMux.Lock() consumers := make([]*BlockUpdateConsumer, 0, len(bl.consumers)) @@ -778,6 +797,7 @@ func (bl *blockListener) GetHeadBlockNumber(_ context.Context) uint64 { } func (bl *blockListener) setHighestBlock(block uint64) { + defer bl.setBlockHeightMetric(metricCanonicalBlockHeight, block) bl.canonicalChainLock.Lock() defer bl.canonicalChainLock.Unlock() bl.highestBlock = block @@ -793,6 +813,9 @@ func (bl *blockListener) checkAndSetHighestBlock(bi *ethrpc.BlockInfoJSONRPC) { bl.highestBlock = block bl.highestBlockSet = true bl.headBlockInfo = bi + // The gauge is bound to the same variable GetHighestBlock reports to event streams, so it is the + // head we are actually tracking rather than a separate sample of it. + bl.setBlockHeightMetric(metricCanonicalBlockHeight, block) } else if block == bl.highestBlock { // Height already known from eth_blockNumber. Store the first full block at that height. bl.headBlockInfo = bi diff --git a/pkg/ethblocklistener/blocklistener_metrics.go b/pkg/ethblocklistener/blocklistener_metrics.go new file mode 100644 index 0000000..c356d49 --- /dev/null +++ b/pkg/ethblocklistener/blocklistener_metrics.go @@ -0,0 +1,92 @@ +// Copyright © 2026 Kaleido, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package ethblocklistener + +import ( + "context" + + "github.com/hyperledger-firefly/common/pkg/i18n" + "github.com/hyperledger-firefly/common/pkg/log" + "github.com/hyperledger-firefly/common/pkg/metric" + "github.com/hyperledger-firefly/evmconnect/internal/msgs" +) + +const ( + metricsSubsystem = "blocklistener" + + // metricTargetBlockHeight is the block height the endpoint we are connected to reports via eth_blockNumber. + // Emitted from queryBlockHeightFromRPC, so it is always the value we last received from the node. + metricTargetBlockHeight = "target_block_height" + // metricCanonicalBlockHeight is the head of the chain this listener is tracking - in full chain tracking + // mode the head of the in-memory canonical chain built from the block filter / newHeads subscription, + // and in light mode the head we dispatch to consumers. It should track the target height very closely. + metricCanonicalBlockHeight = "canonical_block_height" + // metricPollFailures counts the JSON/RPC polls the listen loop makes that failed, labelled by method. + metricPollFailures = "poll_failures_total" + metricLabelPollFailures = "method" +) + +// InitMetrics registers the block listener metrics against the supplied registry. +func (bl *blockListener) InitMetrics(ctx context.Context, registry metric.MetricsRegistry) error { + mm, err := registry.NewMetricsManagerForSubsystem(ctx, metricsSubsystem) + if err != nil { + return i18n.WrapError(ctx, err, msgs.MsgMetricsInitFail, metricsSubsystem) + } + mm.NewGaugeMetric(ctx, metricTargetBlockHeight, "The block height reported by the connected node via eth_blockNumber", false) + mm.NewGaugeMetric(ctx, metricCanonicalBlockHeight, "The block height of the head of the chain tracked by the block listener", false) + mm.NewCounterMetricWithLabels(ctx, metricPollFailures, "The number of block listener JSON/RPC polls that have failed, by method", []string{metricLabelPollFailures}, false) + + bl.metricsLock.Lock() + defer bl.metricsLock.Unlock() + bl.metrics = mm + return nil +} + +func (bl *blockListener) getMetrics() metric.MetricsManager { + bl.metricsLock.RLock() + defer bl.metricsLock.RUnlock() + return bl.metrics +} + +func (bl *blockListener) setBlockHeightMetric(metricName string, blockHeight uint64) { + mm := bl.getMetrics() + if mm == nil { + return + } + mm.SetGaugeMetric(bl.ctx, metricName, float64(blockHeight), nil) +} + +func (bl *blockListener) incPollFailureMetric(method string) { + mm := bl.getMetrics() + if mm == nil { + return + } + mm.IncCounterMetricWithLabels(bl.ctx, metricPollFailures, map[string]string{metricLabelPollFailures: method}, nil) +} + +// refreshTargetBlockHeightMetric queries the node for the height it reports, purely so the target gauge +// stays current. Only needed in full chain tracking mode. +func (bl *blockListener) refreshTargetBlockHeightMetric() { + if bl.getMetrics() == nil { + return // never drive any query of the node when metrics are not enabled + } + if _, err := bl.queryBlockHeightFromRPC(); err != nil { + // Diagnostic only - the failure is recorded on the query failure counter, and the listen loop + // has its own error handling for the chain state + log.L(bl.ctx).Warnf("Failed to refresh target block height: %s", err) + } +} diff --git a/pkg/ethblocklistener/blocklistener_metrics_test.go b/pkg/ethblocklistener/blocklistener_metrics_test.go new file mode 100644 index 0000000..a8e29f8 --- /dev/null +++ b/pkg/ethblocklistener/blocklistener_metrics_test.go @@ -0,0 +1,237 @@ +// Copyright © 2026 Kaleido, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package ethblocklistener + +import ( + "context" + "testing" + "time" + + "github.com/hyperledger-firefly/common/pkg/fftypes" + "github.com/hyperledger-firefly/common/pkg/metric" + "github.com/hyperledger-firefly/evmconnect/mocks/rpcbackendmocks" + "github.com/hyperledger-firefly/evmconnect/pkg/ethrpc" + "github.com/hyperledger-firefly/signer/pkg/ethtypes" + "github.com/hyperledger-firefly/signer/pkg/rpcbackend" + "github.com/hyperledger-firefly/transaction-manager/pkg/ffcapi" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// readGaugeMetric returns the current value of one of our block height gauges, and whether it has +// been reported at all yet +func readGaugeMetric(t *testing.T, registry metric.MetricsRegistry, metricName string) (float64, bool) { + mfs, err := registry.GetGatherer().Gather() + require.NoError(t, err) + fullName := "ff_" + metricsSubsystem + "_" + metricName + for _, mf := range mfs { + if mf.GetName() == fullName { + require.Len(t, mf.GetMetric(), 1) + return mf.GetMetric()[0].GetGauge().GetValue(), true + } + } + return 0, false +} + +// readPollFailureMetric returns the current count of poll failures for the given JSON/RPC method +func readPollFailureMetric(t *testing.T, registry metric.MetricsRegistry, method string) float64 { + mfs, err := registry.GetGatherer().Gather() + require.NoError(t, err) + fullName := "ff_" + metricsSubsystem + "_" + metricPollFailures + for _, mf := range mfs { + if mf.GetName() == fullName { + for _, m := range mf.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == metricLabelPollFailures && l.GetValue() == method { + return m.GetCounter().GetValue() + } + } + } + } + } + return 0 +} + +func waitForGaugeMetric(t *testing.T, registry metric.MetricsRegistry, metricName string, expected float64) { + assert.Eventually(t, func() bool { + v, ok := readGaugeMetric(t, registry, metricName) + return ok && v == expected + }, 5*time.Second, time.Millisecond, "gauge %s did not reach %f", metricName, expected) +} + +func initTestMetrics(t *testing.T, bl *blockListener) metric.MetricsRegistry { + registry := metric.NewPrometheusMetricsRegistry("ut") + require.NoError(t, bl.InitMetrics(context.Background(), registry)) + return registry +} + +func TestBlockListenerMetricsInitFailDuplicateSubsystem(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + registry := metric.NewPrometheusMetricsRegistry("ut") + _, err := registry.NewMetricsManagerForSubsystem(context.Background(), metricsSubsystem) + require.NoError(t, err) + + err = bl.InitMetrics(context.Background(), registry) + assert.Regexp(t, "FF23074", err) + assert.Nil(t, bl.getMetrics()) +} + +func TestBlockListenerMetricsNoopBeforeInit(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + // All emit points no-op, and the target height refresh makes no query at all - the latter + // asserted by done(), as no eth_blockNumber call is mocked + bl.setBlockHeightMetric(metricTargetBlockHeight, 1000) + bl.incPollFailureMetric("eth_blockNumber") + bl.refreshTargetBlockHeightMetric() +} + +func TestBlockListenerMetricsTrackedHeightFromListenerState(t *testing.T) { + _, bl, _, done := newTestBlockListener(t) + defer done() + + registry := initTestMetrics(t, bl) + + // Nothing is reported until we have a height + _, ok := readGaugeMetric(t, registry, metricCanonicalBlockHeight) + assert.False(t, ok) + + // The initial height established at startup from eth_blockNumber + bl.setHighestBlock(1000) + v, ok := readGaugeMetric(t, registry, metricCanonicalBlockHeight) + assert.True(t, ok) + assert.Equal(t, float64(1000), v) + + // Then each block we index that advances the head we are tracking + bl.checkAndSetHighestBlock(ðrpc.BlockInfoJSONRPC{ + Number: ethtypes.HexUint64(1001), + Hash: testBlockHashFor(1001), + }) + v, _ = readGaugeMetric(t, registry, metricCanonicalBlockHeight) + assert.Equal(t, float64(1001), v) + + // Blocks at or below the head we already have don't move it + bl.checkAndSetHighestBlock(ðrpc.BlockInfoJSONRPC{ + Number: ethtypes.HexUint64(999), + Hash: testBlockHashFor(999), + }) + v, _ = readGaugeMetric(t, registry, metricCanonicalBlockHeight) + assert.Equal(t, float64(1001), v) +} + +func TestBlockListenerMetricsTargetBlockHeightQueryFail(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t) + defer done() + + registry := initTestMetrics(t, bl) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(&rpcbackend.RPCError{Message: "pop"}) + + // No retry, and no height reported - just the failure counted, so a node that is failing to answer + // is distinguishable from one reporting a height that isn't moving + bl.refreshTargetBlockHeightMetric() + _, ok := readGaugeMetric(t, registry, metricTargetBlockHeight) + assert.False(t, ok) + assert.Equal(t, float64(1), readPollFailureMetric(t, registry, "eth_blockNumber")) +} + +func TestBlockListenerMetricsFullMode(t *testing.T) { + blockHash1000 := testBlockHashFor(1000) + blockHash1001 := testBlockHashFor(1001) + + ctx, bl, _, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, _ context.CancelFunc) { + conf.BlockPollingInterval = 1 * time.Millisecond + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexIntegerU64(1001) + }) + mockSeedBlockNotFound(mRPC, 1001-uint64(conf.MonitoredHeadLength)+1) + mockNewBlockFilter(mRPC, testBlockFilterID1) + mockFilterChanges(mRPC, testBlockFilterID1, nil, blockHash1001).Once() + mockFilterChangesEmpty(mRPC) + mockBlockByHash(mRPC, 1001, blockHash1001, blockHash1000) + }) + defer done() + + registry := initTestMetrics(t, bl) + + updates := make(chan *ffcapi.BlockHashEvent, 16) + bl.AddConsumer(ctx, &BlockUpdateConsumer{ + ID: fftypes.NewUUID(), + Ctx: ctx, + Updates: updates, + }) + + // The height the node reports, refreshed by the listen loop, and the head of the chain we've built + waitForGaugeMetric(t, registry, metricTargetBlockHeight, 1001) + waitForGaugeMetric(t, registry, metricCanonicalBlockHeight, 1001) +} + +func TestBlockListenerMetricsFullModeFilterFail(t *testing.T) { + _, bl, _, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, _ context.CancelFunc) { + conf.BlockPollingInterval = 1 * time.Millisecond + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexIntegerU64(1001) + }) + mockSeedBlockNotFound(mRPC, 1001-uint64(conf.MonitoredHeadLength)+1) + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_newBlockFilter").Return(&rpcbackend.RPCError{Message: "pop"}) + }) + defer done() + + registry := initTestMetrics(t, bl) + + // Start the loop directly - the filter never establishes, so the listener never marks itself started + bl.checkAndStartListenerLoop() + + // The target height is refreshed ahead of the filter calls, so we can still see the chain moving on + // while the filter is broken - and the failures are counted + waitForGaugeMetric(t, registry, metricTargetBlockHeight, 1001) + assert.Eventually(t, func() bool { + return readPollFailureMetric(t, registry, "eth_newBlockFilter") > 0 + }, 5*time.Second, time.Millisecond) +} + +func TestBlockListenerMetricsLightMode(t *testing.T) { + ctx, bl, _, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, _ context.CancelFunc) { + conf.ChainTrackingMode = ffcapi.ChainTrackingModeLight + conf.BlockPollingInterval = 1 * time.Millisecond + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*ethtypes.HexInteger) = *ethtypes.NewHexIntegerU64(2000) + }) + mockNewBlockFilter(mRPC, testBlockFilterID1) + mockFilterChangesEmpty(mRPC) + }) + defer done() + + registry := initTestMetrics(t, bl) + + updates := make(chan *ffcapi.BlockHashEvent, 16) + bl.AddConsumer(ctx, &BlockUpdateConsumer{ + ID: fftypes.NewUUID(), + Ctx: ctx, + Updates: updates, + }) + + // In light mode there is no canonical chain, so the head we dispatch is the height we track + waitForGaugeMetric(t, registry, metricTargetBlockHeight, 2000) + waitForGaugeMetric(t, registry, metricCanonicalBlockHeight, 2000) +}