From 0e8e3e6d18177e7ad8a8844f57a36fb44aa5e19f Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:35:24 +0200 Subject: [PATCH 1/7] feat(verifier): emit block receive delay and SLO threshold metrics Record block_receive_delay_seconds for headers received from the subscription (clamped at 0, not for backfilled headers) and initialize the block time and receive delay SLO gauges at verifier start. Fixes #44 Co-Authored-By: Claude Sonnet 5.5 --- README.md | 6 +-- pkg/exporters/verifier/verifier.go | 3 ++ pkg/exporters/verifier/verifier_test.go | 54 +++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index ada72ce..0bbc329 100644 --- a/README.md +++ b/README.md @@ -194,7 +194,7 @@ The following metrics are exposed: ### `ev_metrics_block_time_slo_seconds` - **Type**: Gauge - **Labels**: `chain_id`, `quantile` -- **Description**: SLO thresholds for block time (not yet emitted, see #44) +- **Description**: SLO thresholds for block time - **Values**: - `0.5`: 2.0s - `0.9`: 3.0s @@ -204,13 +204,13 @@ The following metrics are exposed: ### `ev_metrics_block_receive_delay_seconds` - **Type**: Histogram - **Labels**: `chain_id` -- **Description**: Delay between block creation and reception with histogram buckets (not yet emitted, see #44) +- **Description**: Delay between block creation (`header.Time`) and reception with histogram buckets. Only headers received from the subscription are observed (not backfilled ones), and negative values from clock skew are recorded as 0. `header.Time` has 1-second resolution, so the value is only accurate to about ±1s. - **Buckets**: 0.1, 0.25, 0.5, 1.0, 2.0, 3.0, 5.0, 10.0, 15.0, 30.0, 60.0 seconds ### `ev_metrics_block_receive_delay_slo_seconds` - **Type**: Gauge - **Labels**: `chain_id`, `quantile` -- **Description**: SLO thresholds for block receive delay (not yet emitted, see #44) +- **Description**: SLO thresholds for block receive delay - **Values**: - `0.5`: 1.0s - `0.9`: 3.0s diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 71e2a6d..4df290f 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -121,6 +121,8 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error // pre-initialize submission metrics so both blob types are always visible // in Prometheus output, even before any block has been fully processed. m.InitializeSubmissionMetrics(e.chainID) + m.InitializeBlockTimeSloThresholds(e.chainID) + m.InitializeBlockReceiveDelaySloThresholds(e.chainID) // ticker to refresh submission duration metric every 10 seconds refreshTicker := time.NewTicker(10 * time.Second) @@ -168,6 +170,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error // record block arrival time for millisecond precision arrivalTime := time.Now() m.RecordBlockTime(e.chainID, arrivalTime) + m.RecordBlockReceiveDelay(e.chainID, max(0, arrivalTime.Sub(time.Unix(int64(header.Time), 0)))) height := header.Number.Uint64() e.logger.Debug(). diff --git a/pkg/exporters/verifier/verifier_test.go b/pkg/exporters/verifier/verifier_test.go index cc965c3..5008cc5 100644 --- a/pkg/exporters/verifier/verifier_test.go +++ b/pkg/exporters/verifier/verifier_test.go @@ -173,6 +173,24 @@ func (h *harness) metric(t *testing.T, name string) float64 { return sum } +func (h *harness) sum(t *testing.T, name string) float64 { + t.Helper() + families, err := h.reg.Gather() + if err != nil { + t.Fatal(err) + } + var sum float64 + for _, mf := range families { + if mf.GetName() != "test_"+name { + continue + } + for _, m := range mf.GetMetric() { + sum += m.GetHistogram().GetSampleSum() + } + } + return sum +} + func eventually(t *testing.T, msg string, cond func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) @@ -287,6 +305,42 @@ func TestExportMetricsBackfillsGap(t *testing.T) { } } +func TestExportMetricsRecordsReceiveDelayAndSloGauges(t *testing.T) { + h := newHarness(1, nil, 0, time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error) + go func() { done <- h.e.ExportMetrics(ctx, h.m) }() + defer func() { + cancel() + if err := <-done; err != nil { + t.Error(err) + } + }() + + eventually(t, "not subscribed", func() bool { + h.heads.mu.Lock() + defer h.heads.mu.Unlock() + return h.heads.ch != nil + }) + header := newHeader(1) + header.Time = uint64(time.Now().Add(-3 * time.Second).Unix()) + h.heads.ch <- header + + eventually(t, "receive delay not recorded", func() bool { return h.metric(t, "block_receive_delay_seconds") == 1 }) + if got := h.metric(t, "block_receive_delay_seconds"); got != 1 { + t.Fatalf("expected 1 receive delay observation, got %v", got) + } + if got := h.sum(t, "block_receive_delay_seconds"); got < 2 || got > 5 { + t.Fatalf("expected a delay of about 3s, got %v", got) + } + if got := h.metric(t, "block_time_slo_seconds"); got != 2+3+4+5 { + t.Fatalf("unexpected block time SLO gauges sum %v", got) + } + if got := h.metric(t, "block_receive_delay_slo_seconds"); got != 1+3+5+10 { + t.Fatalf("unexpected receive delay SLO gauges sum %v", got) + } +} + func TestBackfillCapsGap(t *testing.T) { h := newHarness(1, nil) queue := make(chan job, maxBackfill+10) From dc31a6d901aebb2e8ace2e17060d150d23c91315 Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:36:35 +0200 Subject: [PATCH 2/7] fix(security): redact credentials and paths from endpoint labels and logs Endpoint labels and log fields now show only scheme://host[:port]. Errors are redacted before logging, and endpoints that redact to the same value get a # suffix so their labels stay distinct. Fixes #49 Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 + cmd/monitor.go | 3 +- pkg/exporters/balance/balance.go | 25 ++++--- pkg/exporters/balance/balance_test.go | 58 ++++++++++++++ pkg/exporters/drift/drift.go | 47 +++++++----- pkg/exporters/drift/drift_test.go | 73 ++++++++++++++++++ pkg/utils/endpoint.go | 104 ++++++++++++++++++++++++++ pkg/utils/endpoint_test.go | 59 +++++++++++++++ 8 files changed, 339 insertions(+), 32 deletions(-) create mode 100644 pkg/utils/endpoint.go create mode 100644 pkg/utils/endpoint_test.go diff --git a/README.md b/README.md index ada72ce..24e96e2 100644 --- a/README.md +++ b/README.md @@ -241,6 +241,8 @@ When `--evm-rpc-url` is provided: When `--reference-node` and `--full-nodes` are provided: +Endpoint labels (here and in the balance metrics below) and logs show only `scheme://host[:port]`, so credentials in endpoint URLs (userinfo, path, query) are not exposed. Endpoints that redact to the same value get a `#` suffix to stay distinct. + ### `ev_metrics_reference_block_height` - **Type**: Gauge - **Labels**: `chain_id`, `endpoint` diff --git a/cmd/monitor.go b/cmd/monitor.go index 7d6dbc2..847b85c 100644 --- a/cmd/monitor.go +++ b/cmd/monitor.go @@ -19,6 +19,7 @@ import ( "github.com/evstack/ev-metrics/pkg/exporters/jsonrpc" "github.com/evstack/ev-metrics/pkg/exporters/verifier" "github.com/evstack/ev-metrics/pkg/metrics" + "github.com/evstack/ev-metrics/pkg/utils" coreda "github.com/evstack/ev-node/pkg/da/types" "github.com/rs/zerolog" "github.com/spf13/cobra" @@ -255,7 +256,7 @@ func monitorAndExportMetrics(_ *cobra.Command, _ []string) error { logger.Info(). Strs("addresses", addressList). - Strs("rpc_urls", rpcUrls). + Strs("rpc_urls", utils.RedactEndpoints(rpcUrls)). Int("scrape_interval", flags.balanceScrapeInterval). Msg("balance monitoring enabled") } diff --git a/pkg/exporters/balance/balance.go b/pkg/exporters/balance/balance.go index b7b2198..cb8f84b 100644 --- a/pkg/exporters/balance/balance.go +++ b/pkg/exporters/balance/balance.go @@ -27,6 +27,7 @@ func NewMetricsExporter( chainID: chainID, addresses: addresses, endpoints: endpoints, + labels: utils.RedactEndpoints(endpoints), scrapeInterval: scrapeInterval, rpcTimeout: defaultRPCTimeout, logger: logger.With().Str("component", "balance_monitor").Logger(), @@ -38,6 +39,7 @@ type exporter struct { chainID string addresses []string endpoints []string + labels []string scrapeInterval int rpcTimeout time.Duration logger zerolog.Logger @@ -60,7 +62,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error e.logger.Info(). Strs("addresses", e.addresses). - Strs("endpoints", e.endpoints). + Strs("endpoints", e.labels). Int("scrape_interval_sec", e.scrapeInterval). Msg("starting balance monitoring") @@ -82,16 +84,17 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, clients []*cosmos.Client, address string) { // try each endpoint until one succeeds for i, endpoint := range e.endpoints { + label := e.labels[i] if clients[i] == nil { client, err := cosmos.NewClient(endpoint, e.logger) if err != nil { e.logger.Warn(). - Err(err). - Str("endpoint", endpoint). + Err(utils.RedactError(err, endpoint)). + Str("endpoint", label). Str("address", address). Msg("failed to create client, trying next endpoint") - m.RecordConsensusRpcEndpointAvailability(e.chainID, endpoint, false) - m.RecordConsensusRpcEndpointError(e.chainID, endpoint, utils.CategorizeError(err)) + m.RecordConsensusRpcEndpointAvailability(e.chainID, label, false) + m.RecordConsensusRpcEndpointError(e.chainID, label, utils.CategorizeError(err)) continue } clients[i] = client @@ -103,17 +106,17 @@ func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, clients if err != nil { e.logger.Warn(). - Err(err). - Str("endpoint", endpoint). + Err(utils.RedactError(err, endpoint)). + Str("endpoint", label). Str("address", address). Msg("failed to query balance, trying next endpoint") - m.RecordConsensusRpcEndpointAvailability(e.chainID, endpoint, false) - m.RecordConsensusRpcEndpointError(e.chainID, endpoint, utils.CategorizeError(err)) + m.RecordConsensusRpcEndpointAvailability(e.chainID, label, false) + m.RecordConsensusRpcEndpointError(e.chainID, label, utils.CategorizeError(err)) continue } // success - record availability - m.RecordConsensusRpcEndpointAvailability(e.chainID, endpoint, true) + m.RecordConsensusRpcEndpointAvailability(e.chainID, label, true) // record balances for each denom for _, coin := range balances { @@ -135,7 +138,7 @@ func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, clients Str("address", address). Str("denom", coin.Denom). Float64("balance", balanceFloat). - Str("endpoint", endpoint). + Str("endpoint", label). Msg("recorded account balance") } diff --git a/pkg/exporters/balance/balance_test.go b/pkg/exporters/balance/balance_test.go index ab83255..dae797b 100644 --- a/pkg/exporters/balance/balance_test.go +++ b/pkg/exporters/balance/balance_test.go @@ -1,6 +1,7 @@ package balance import ( + "bytes" "context" "net" "sync/atomic" @@ -97,3 +98,60 @@ func TestExportMetrics_FailoverAndClientReuse(t *testing.T) { require.Equal(t, map[string]float64{"addr1": 100, "addr2": 200}, balances()) require.Equal(t, int32(1), lis.accepted.Load(), "both addresses must share one connection") } + +func TestExportMetrics_EndpointSecretsNeverExposed(t *testing.T) { + const secret = "SECRET_API_KEY" + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + srv := grpc.NewServer() + banktypes.RegisterQueryServer(srv, &bankServer{}) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + dead, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + deadAddr := dead.Addr().String() + require.NoError(t, dead.Close()) + + var logs bytes.Buffer + logger := zerolog.New(&logs).Level(zerolog.DebugLevel) + reg := prometheus.NewRegistry() + m := metrics.NewWithRegistry("test", reg) + endpoints := []string{"http://" + deadAddr + "/v1/" + secret, "http://" + lis.Addr().String() + "/v1/" + secret} + e := NewMetricsExporter("chain", []string{"addr1"}, endpoints, 60, logger).(*exporter) + e.rpcTimeout = time.Second + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = e.ExportMetrics(ctx, m) + }() + + require.Eventually(t, func() bool { + families, err := reg.Gather() + require.NoError(t, err) + for _, mf := range families { + if mf.GetName() == "test_account_balance" { + return true + } + } + return false + }, 3*time.Second, 20*time.Millisecond) + cancel() + <-done + + families, err := reg.Gather() + require.NoError(t, err) + labels := map[string]bool{} + for _, mf := range families { + for _, metric := range mf.GetMetric() { + for _, l := range metric.GetLabel() { + require.NotContains(t, l.GetValue(), secret, "metric %s label %s", mf.GetName(), l.GetName()) + labels[l.GetValue()] = true + } + } + } + require.True(t, labels["http://"+deadAddr]) + require.NotContains(t, logs.String(), secret) +} diff --git a/pkg/exporters/drift/drift.go b/pkg/exporters/drift/drift.go index b28dd59..5b4b099 100644 --- a/pkg/exporters/drift/drift.go +++ b/pkg/exporters/drift/drift.go @@ -18,10 +18,13 @@ const defaultRPCTimeout = 5 * time.Second // NewMetricsExporter creates and returns a metrics exporter for monitoring block height drift across multiple nodes. func NewMetricsExporter(chainID, referenceNode string, fullNodes []string, pollingInterval int, logger zerolog.Logger) metrics.Exporter { + labels := utils.RedactEndpoints(append([]string{referenceNode}, fullNodes...)) return &exporter{ chainID: chainID, referenceNode: referenceNode, + referenceLabel: labels[0], fullNodes: fullNodes, + fullNodeLabels: labels[1:], pollingInterval: pollingInterval, rpcTimeout: defaultRPCTimeout, logger: logger.With().Str("component", "drift_monitor").Logger(), @@ -32,7 +35,9 @@ func NewMetricsExporter(chainID, referenceNode string, fullNodes []string, polli type exporter struct { chainID string referenceNode string + referenceLabel string fullNodes []string + fullNodeLabels []string pollingInterval int rpcTimeout time.Duration logger zerolog.Logger @@ -40,9 +45,11 @@ type exporter struct { } // node holds a lazily dialed client for one endpoint. Polls never overlap, so a node is -// only ever used by one goroutine at a time. +// only ever used by one goroutine at a time. url is only for dialing; label is the redacted +// form used in metrics and logs. type node struct { url string + label string client *ethclient.Client } @@ -59,10 +66,10 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error ticker := time.NewTicker(time.Duration(e.pollingInterval) * time.Second) defer ticker.Stop() - ref := &node{url: e.referenceNode} + ref := &node{url: e.referenceNode, label: e.referenceLabel} fullNodes := make([]*node, len(e.fullNodes)) for i, url := range e.fullNodes { - fullNodes[i] = &node{url: url} + fullNodes[i] = &node{url: url, label: e.fullNodeLabels[i]} } defer func() { ref.close() @@ -72,8 +79,8 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error }() e.logger.Info(). - Str("reference_node", e.referenceNode). - Strs("full_nodes", e.fullNodes). + Str("reference_node", e.referenceLabel). + Strs("full_nodes", e.fullNodeLabels). Int("polling_interval_sec", e.pollingInterval). Msg("starting node drift monitoring") @@ -93,21 +100,21 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error func (e *exporter) poll(ctx context.Context, m *metrics.Metrics, ref *node, fullNodes []*node) { refHeight, err := e.blockHeight(ctx, ref) if err != nil { - e.logger.Error().Err(err).Str("endpoint", ref.url).Msg("failed to get reference node block height") - m.RecordEndpointAvailability(e.chainID, ref.url, false) - m.RecordEndpointError(e.chainID, ref.url, utils.CategorizeError(err)) + e.logger.Error().Err(err).Str("endpoint", ref.label).Msg("failed to get reference node block height") + m.RecordEndpointAvailability(e.chainID, ref.label, false) + m.RecordEndpointError(e.chainID, ref.label, utils.CategorizeError(err)) return } - m.RecordEndpointAvailability(e.chainID, ref.url, true) - m.RecordReferenceBlockHeight(e.chainID, ref.url, refHeight) + m.RecordEndpointAvailability(e.chainID, ref.label, true) + m.RecordReferenceBlockHeight(e.chainID, ref.label, refHeight) // Also update time since last block metric when we see new blocks via polling // This ensures the metric stays accurate even if WebSocket subscription fails if refHeight > e.lastHeight { m.UpdateLastBlockTime(e.chainID, time.Now()) e.lastHeight = refHeight } - e.logger.Debug().Uint64("height", refHeight).Str("endpoint", ref.url).Msg("recorded reference node height") + e.logger.Debug().Uint64("height", refHeight).Str("endpoint", ref.label).Msg("recorded reference node height") var wg sync.WaitGroup for _, n := range fullNodes { @@ -123,21 +130,21 @@ func (e *exporter) poll(ctx context.Context, m *metrics.Metrics, ref *node, full func (e *exporter) pollFullNode(ctx context.Context, m *metrics.Metrics, n *node, refHeight uint64) { currentHeight, err := e.blockHeight(ctx, n) if err != nil { - e.logger.Error().Err(err).Str("endpoint", n.url).Msg("failed to get full node block height") - m.RecordEndpointAvailability(e.chainID, n.url, false) - m.RecordEndpointError(e.chainID, n.url, utils.CategorizeError(err)) + e.logger.Error().Err(err).Str("endpoint", n.label).Msg("failed to get full node block height") + m.RecordEndpointAvailability(e.chainID, n.label, false) + m.RecordEndpointError(e.chainID, n.label, utils.CategorizeError(err)) return } - m.RecordEndpointAvailability(e.chainID, n.url, true) - m.RecordCurrentBlockHeight(e.chainID, n.url, currentHeight) - m.RecordBlockHeightDrift(e.chainID, n.url, refHeight, currentHeight) + m.RecordEndpointAvailability(e.chainID, n.label, true) + m.RecordCurrentBlockHeight(e.chainID, n.label, currentHeight) + m.RecordBlockHeightDrift(e.chainID, n.label, refHeight, currentHeight) e.logger.Debug(). Uint64("ref_height", refHeight). Uint64("target_height", currentHeight). Int64("drift", int64(refHeight)-int64(currentHeight)). - Str("endpoint", n.url). + Str("endpoint", n.label). Msg("recorded full node height and drift") } @@ -150,7 +157,7 @@ func (e *exporter) blockHeight(ctx context.Context, n *node) (uint64, error) { if n.client == nil { client, err := ethclient.DialContext(ctx, n.url) if err != nil { - return 0, fmt.Errorf("failed to connect to %s: %w", n.url, err) + return 0, utils.RedactError(fmt.Errorf("failed to connect to %s: %w", n.label, err), n.url) } n.client = client } @@ -158,7 +165,7 @@ func (e *exporter) blockHeight(ctx context.Context, n *node) (uint64, error) { height, err := n.client.BlockNumber(ctx) if err != nil { n.close() - return 0, fmt.Errorf("failed to get block number from %s: %w", n.url, err) + return 0, utils.RedactError(fmt.Errorf("failed to get block number from %s: %w", n.label, err), n.url) } return height, nil diff --git a/pkg/exporters/drift/drift_test.go b/pkg/exporters/drift/drift_test.go index 416c7a0..1fcefb2 100644 --- a/pkg/exporters/drift/drift_test.go +++ b/pkg/exporters/drift/drift_test.go @@ -1,17 +1,20 @@ package drift import ( + "bytes" "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" + "sync" "sync/atomic" "testing" "time" "github.com/evstack/ev-metrics/pkg/metrics" + "github.com/evstack/ev-metrics/pkg/utils" "github.com/prometheus/client_golang/prometheus" "github.com/rs/zerolog" "github.com/stretchr/testify/require" @@ -147,3 +150,73 @@ func TestExportMetrics_HungFullNodeDoesNotBlockOthers(t *testing.T) { return ok && avail == 0 }, 3*time.Second, 20*time.Millisecond) } + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +func TestExportMetrics_EndpointSecretsNeverExposed(t *testing.T) { + const secret = "SECRET_API_KEY" + var height atomic.Uint64 + height.Store(10) + ref := newBlockNumberServer(t, &height) + good := newBlockNumberServer(t, &height) + dead := httptest.NewServer(http.NotFoundHandler()) + deadURL := dead.URL + dead.Close() + + refURL := ref.URL + "/v2/" + secret + goodURL := "http://user:" + secret + "@" + good.Listener.Addr().String() + "/v2/" + secret + "?token=" + secret + deadSecretURL := deadURL + "/v2/" + secret + + var logs syncBuffer + logger := zerolog.New(&logs).Level(zerolog.DebugLevel) + reg := prometheus.NewRegistry() + m := metrics.NewWithRegistry("test", reg) + e := NewMetricsExporter("chain", refURL, []string{goodURL, deadSecretURL}, 60, logger).(*exporter) + e.rpcTimeout = time.Second + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = e.ExportMetrics(ctx, m) + }() + + deadLabel := utils.RedactEndpoint(deadURL) + require.Eventually(t, func() bool { + avail, ok := gaugeValue(t, reg, "test_endpoint_availability", deadLabel) + return ok && avail == 0 + }, 3*time.Second, 10*time.Millisecond) + require.Eventually(t, func() bool { + _, ok := gaugeValue(t, reg, "test_block_height_drift", utils.RedactEndpoint(goodURL)) + return ok + }, 3*time.Second, 10*time.Millisecond) + cancel() + <-done + + families, err := reg.Gather() + require.NoError(t, err) + for _, mf := range families { + for _, metric := range mf.GetMetric() { + for _, l := range metric.GetLabel() { + require.NotContains(t, l.GetValue(), secret, "metric %s label %s", mf.GetName(), l.GetName()) + } + } + } + require.NotContains(t, logs.String(), secret) + require.Contains(t, logs.String(), "failed to get block number") +} diff --git a/pkg/utils/endpoint.go b/pkg/utils/endpoint.go new file mode 100644 index 0000000..3dde52e --- /dev/null +++ b/pkg/utils/endpoint.go @@ -0,0 +1,104 @@ +package utils + +import ( + "net/url" + "sort" + "strconv" + "strings" +) + +// InvalidEndpoint replaces endpoints that cannot be parsed, so that their content is never echoed. +const InvalidEndpoint = "invalid-endpoint" + +// RedactEndpoint reduces an endpoint URL to scheme://host[:port], dropping userinfo, path, query +// and fragment, which commonly carry credentials. Input without a scheme is treated as host[:port]. +func RedactEndpoint(raw string) string { + if !strings.Contains(raw, "://") { + // url.Parse reads "host:9090" as scheme "host", so strip by hand + if i := strings.IndexAny(raw, "/?#"); i >= 0 { + raw = raw[:i] + } + if i := strings.LastIndexByte(raw, '@'); i >= 0 { + raw = raw[i+1:] + } + return raw + } + + u, err := url.Parse(raw) + if err != nil || u.Scheme == "" || u.Host == "" { + return InvalidEndpoint + } + return u.Scheme + "://" + u.Host +} + +// RedactEndpoints redacts each endpoint and keeps the results distinct: every repeat of an already +// seen redacted form gets "#" appended, where index is its position in raws. +func RedactEndpoints(raws []string) []string { + out := make([]string, len(raws)) + seen := make(map[string]bool, len(raws)) + for i, raw := range raws { + r := RedactEndpoint(raw) + if seen[r] { + r += "#" + strconv.Itoa(i) + } + seen[r] = true + out[i] = r + } + return out +} + +type redactedError struct { + msg string + err error +} + +func (e *redactedError) Error() string { return e.msg } +func (e *redactedError) Unwrap() error { return e.err } + +// RedactError returns err with every occurrence of the raw endpoints, and of their userinfo and +// path/query parts, removed from the message. The original error stays reachable through +// errors.Is and errors.As. +func RedactError(err error, raws ...string) error { + if err == nil { + return nil + } + + repl := map[string]string{} + for _, raw := range raws { + if raw == "" { + continue + } + repl[raw] = RedactEndpoint(raw) + u, perr := url.Parse(raw) + if perr != nil { + continue + } + if u.User != nil { + repl[u.User.String()+"@"] = "" + } + if p := u.RequestURI(); u.Host != "" && len(p) > 1 { + repl[p] = "" + } + if p := u.EscapedPath(); len(p) > 1 { + repl[p] = "" + } + } + + // longest first: Replacer prefers earlier pairs when several match at one position + olds := make([]string, 0, len(repl)) + for old := range repl { + olds = append(olds, old) + } + sort.Slice(olds, func(i, j int) bool { + if len(olds[i]) != len(olds[j]) { + return len(olds[i]) > len(olds[j]) + } + return olds[i] < olds[j] + }) + pairs := make([]string, 0, 2*len(olds)) + for _, old := range olds { + pairs = append(pairs, old, repl[old]) + } + + return &redactedError{msg: strings.NewReplacer(pairs...).Replace(err.Error()), err: err} +} diff --git a/pkg/utils/endpoint_test.go b/pkg/utils/endpoint_test.go new file mode 100644 index 0000000..802df49 --- /dev/null +++ b/pkg/utils/endpoint_test.go @@ -0,0 +1,59 @@ +package utils + +import ( + "errors" + "fmt" + "io/fs" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRedactEndpoint(t *testing.T) { + tests := []struct{ in, want string }{ + {"https://host/v2/KEY?x=1", "https://host"}, + {"https://user:pass@host:8545/path", "https://host:8545"}, + {"http://host:9090", "http://host:9090"}, + {"host:9090", "host:9090"}, + {"host:9090/v2/KEY?x=1", "host:9090"}, + {"user:pass@host:9090", "host:9090"}, + {"https://host#frag", "https://host"}, + {"https://SECRET_KEY%zz@/path", InvalidEndpoint}, + {"https:///v2/SECRET_KEY", InvalidEndpoint}, + {"://SECRET_KEY", InvalidEndpoint}, + } + for _, tt := range tests { + t.Run(tt.in, func(t *testing.T) { + got := RedactEndpoint(tt.in) + require.Equal(t, tt.want, got) + require.NotContains(t, got, "SECRET_KEY") + }) + } +} + +func TestRedactEndpoints(t *testing.T) { + got := RedactEndpoints([]string{"https://gw/a/KEY1", "https://gw/b/KEY2", "https://other/x", "https://gw/c/KEY3"}) + require.Equal(t, []string{"https://gw", "https://gw#1", "https://other", "https://gw#3"}, got) +} + +func TestRedactError(t *testing.T) { + raw := "https://user:pass@host:8545/v2/SECRET_KEY?token=SECRET_TOK" + err := fmt.Errorf("failed to connect to %s: %w", raw, fs.ErrNotExist) + err = fmt.Errorf("dial target user:pass@host:8545 path /v2/SECRET_KEY: %w", err) + + red := RedactError(err, raw) + msg := red.Error() + for _, secret := range []string{"pass", "SECRET_KEY", "SECRET_TOK"} { + require.False(t, strings.Contains(msg, secret), "%q leaked in %q", secret, msg) + } + require.Contains(t, msg, "https://host:8545") + + require.ErrorIs(t, red, fs.ErrNotExist) + var pe *fs.PathError + pathErr := &fs.PathError{Op: "open", Path: raw, Err: fs.ErrNotExist} + require.True(t, errors.As(RedactError(pathErr, raw), &pe)) + require.NotContains(t, RedactError(pathErr, raw).Error(), "SECRET_KEY") + + require.NoError(t, RedactError(nil, raw)) +} From b56a307fb8f5cac3cc4137e6835bd582f626c769 Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:37:12 +0200 Subject: [PATCH 3/7] fix(cosmos): drop userinfo and query from the gRPC target so credentials never reach logs Follow-up to the endpoint redaction for #49: the cosmos client logs its dial target at debug level and in connect errors, and parseEndpoint kept userinfo and a bare query. Refs #49 Co-Authored-By: Claude Sonnet 5.5 --- internal/clients/cosmos/client.go | 8 ++++++-- internal/clients/cosmos/client_test.go | 3 +++ 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/internal/clients/cosmos/client.go b/internal/clients/cosmos/client.go index 0c410be..4dada1d 100644 --- a/internal/clients/cosmos/client.go +++ b/internal/clients/cosmos/client.go @@ -25,13 +25,17 @@ type Client struct { // parseEndpoint converts an endpoint URL into a gRPC "host:port" target and reports whether TLS // should be used. https:// endpoints use TLS and default to port 443; http:// and bare host:port -// endpoints are plaintext. Any path is dropped. +// endpoints are plaintext. Userinfo, path, query and fragment are dropped, so credentials in the URL +// never reach the dial target, the logs or error messages. func parseEndpoint(endpoint string) (target string, useTLS bool) { useTLS = strings.HasPrefix(endpoint, "https://") target = strings.TrimPrefix(strings.TrimPrefix(endpoint, "https://"), "http://") - if i := strings.IndexByte(target, '/'); i >= 0 { + if i := strings.IndexAny(target, "/?#"); i >= 0 { target = target[:i] } + if i := strings.LastIndexByte(target, '@'); i >= 0 { + target = target[i+1:] + } if useTLS { if _, _, err := net.SplitHostPort(target); err != nil { target = net.JoinHostPort(strings.Trim(target, "[]"), "443") diff --git a/internal/clients/cosmos/client_test.go b/internal/clients/cosmos/client_test.go index 2838b5c..7f8737d 100644 --- a/internal/clients/cosmos/client_test.go +++ b/internal/clients/cosmos/client_test.go @@ -17,6 +17,9 @@ func TestParseEndpoint(t *testing.T) { {"https://h:443/path", "h:443", true}, {"https://h:9443", "h:9443", true}, {"h:9090", "h:9090", false}, + {"https://user:pass@h:9443/p", "h:9443", true}, + {"http://user:pass@h:9090", "h:9090", false}, + {"https://h?key=SECRET", "h:443", true}, } for _, tt := range tests { t.Run(tt.endpoint, func(t *testing.T) { From 0f06b29686127b8690ad975541c0e0b70f705052 Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:37:22 +0200 Subject: [PATCH 4/7] feat(verifier): add self-observability metrics and a meaningful /health Add verifier queue length, in-flight blocks, WebSocket reconnect, last header timestamp and backfill counters. /health now returns 503 until the header subscription is established, while it is down, and (with --health-max-header-age) when no header arrived within the limit. Fixes #45 Fixes #46 Co-Authored-By: Claude Sonnet 5.5 --- README.md | 42 +++++++ cmd/monitor.go | 5 +- pkg/exporters/verifier/verifier.go | 45 +++++--- pkg/exporters/verifier/verifier_test.go | 104 +++++++++++++++++- pkg/metrics/metrics.go | 139 ++++++++++++++++++++++++ pkg/metrics/metrics_test.go | 78 +++++++++++++ pkg/metrics/server.go | 26 +++-- pkg/metrics/server_test.go | 64 ++++++++++- 8 files changed, 477 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 0bbc329..67a7a50 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,7 @@ Metrics will be available at `http://localhost:2112/metrics` - `--balance.consensus-rpc-urls`: Comma-separated consensus RPC URLs for balance queries (required if balance.addresses is set) - `--balance.scrape-interval`: Balance check scrape interval in seconds (default: 30) - `--verifier.workers`: Number of concurrent workers for block verification (default: 50) +- `--health-max-header-age`: Seconds without a new block header after which `/health` returns 503 (default: 0, which disables the age check so only the subscription state matters). See [Health Endpoint](#health-endpoint). - `--verbose`: Enable verbose (debug level) logging (default: false) - `--log-format`: Log output format, `console` (human-readable) or `json` (one JSON object per line, for Loki/ELK ingestion) (default: `console`) @@ -217,6 +218,47 @@ The following metrics are exposed: - `0.95`: 5.0s - `0.99`: 10.0s +### Verifier Self-Observability Metrics + +### `ev_metrics_verifier_queue_length` +- **Type**: Gauge +- **Labels**: `chain_id` +- **Description**: Number of blocks waiting in the verifier queue, sampled every 10 seconds. A value that stays near the queue capacity (2 x `--verifier.workers`) means the verifier is falling behind. + +### `ev_metrics_verifier_inflight_blocks` +- **Type**: Gauge +- **Labels**: `chain_id` +- **Description**: Blocks between their first verification attempt and being verified or running out of retries, including blocks waiting on a retry timer. A block re-queued after the cooldown is counted again. + +### `ev_metrics_ws_reconnects_total` +- **Type**: Counter +- **Labels**: `chain_id` +- **Description**: Number of times the block header WebSocket subscription was re-established after dropping. + +### `ev_metrics_last_header_received_timestamp` +- **Type**: Gauge +- **Labels**: `chain_id` +- **Description**: Unix timestamp in seconds of the last block header received from the subscription (backfilled headers are not counted). It is 0 until the first header arrives. + +### `ev_metrics_verifier_backfilled_blocks_total` +- **Type**: Counter +- **Labels**: `chain_id` +- **Description**: Blocks enqueued after a gap in the header subscription (for example after a reconnect). + +### `ev_metrics_verifier_backfill_dropped_total` +- **Type**: Counter +- **Labels**: `chain_id` +- **Description**: Blocks skipped because a subscription gap was larger than the 1000-block backfill cap. The oldest blocks are dropped. + +### Health Endpoint + +`GET /health` returns `200 OK`, or `503` with a short plain-text reason when: +- the block header subscription has not been established yet (`not ready`), +- the subscription is down (it is re-established automatically with backoff), or +- `--health-max-header-age` is greater than 0 and no header has arrived within that many seconds (measured from the first subscription until a header arrives). + +Use `/health` as a **readiness** probe, not a liveness probe. With the age check enabled, a halted chain also fails it, and restarting the monitor would not fix that. + ### JSON-RPC Monitoring Metrics When `--evm-rpc-url` is provided: diff --git a/cmd/monitor.go b/cmd/monitor.go index 7d6dbc2..b883754 100644 --- a/cmd/monitor.go +++ b/cmd/monitor.go @@ -45,6 +45,7 @@ const ( flagBalanceRpcUrls = "balance.consensus-rpc-urls" flagBalanceScrapeInterval = "balance.scrape-interval" flagVerifierWorkers = "verifier.workers" + flagHealthMaxHeaderAge = "health-max-header-age" metricsPath = "/metrics" ) @@ -72,6 +73,7 @@ type flagValues struct { balanceRpcUrls string balanceScrapeInterval int verifierWorkers int + healthMaxHeaderAge int } func NewMonitorCmd() *cobra.Command { @@ -106,6 +108,7 @@ func NewMonitorCmd() *cobra.Command { cmd.Flags().StringVar(&flags.balanceRpcUrls, flagBalanceRpcUrls, "", "Comma-separated consensus rpc urls for balance queries (required if balance.addresses is set)") cmd.Flags().IntVar(&flags.balanceScrapeInterval, flagBalanceScrapeInterval, 30, "Balance check scrape interval in seconds (default: 30)") cmd.Flags().IntVar(&flags.verifierWorkers, flagVerifierWorkers, 50, "Number of concurrent workers for block verification (default: 50)") + cmd.Flags().IntVar(&flags.healthMaxHeaderAge, flagHealthMaxHeaderAge, 0, "/health fails when no block header has arrived for this many seconds (0 = disabled, only subscription state is checked)") if err := cmd.MarkFlagRequired(flagHeaderNS); err != nil { panic(err) @@ -260,7 +263,7 @@ func monitorAndExportMetrics(_ *cobra.Command, _ []string) error { Msg("balance monitoring enabled") } - err = metrics.StartServer(ctx, metricsPath, flags.port, logger, exporters...) + err = metrics.StartServer(ctx, metricsPath, flags.port, time.Duration(flags.healthMaxHeaderAge)*time.Second, logger, exporters...) // context.Canceled (shutdown) and context.DeadlineExceeded (--duration) are clean exits if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 4df290f..556e8cd 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -52,6 +52,8 @@ var defaultRetryIntervals = []time.Duration{ const ( // defaultCooldown is how long to wait before re-verifying a block whose retries were exhausted. defaultCooldown = 5 * time.Minute + // defaultReconnectBackoff is the initial delay before re-subscribing; it doubles up to a 60s cap. + defaultReconnectBackoff = 5 * time.Second // maxBackfill caps how many heights are fetched after a gap in the header subscription. maxBackfill = 1000 ) @@ -67,16 +69,17 @@ func NewMetricsExporter( logger zerolog.Logger, ) metrics.Exporter { return &exporter{ - evnodeClient: evnodeClient, - evmClient: evmClient, - celestiaClient: celestiaClient, - headerNS: headerNS, - dataNS: dataNS, - chainID: chainID, - workers: workers, - retryIntervals: defaultRetryIntervals, - cooldown: defaultCooldown, - logger: logger.With().Str("component", "verification_monitor").Logger(), + evnodeClient: evnodeClient, + evmClient: evmClient, + celestiaClient: celestiaClient, + headerNS: headerNS, + dataNS: dataNS, + chainID: chainID, + workers: workers, + retryIntervals: defaultRetryIntervals, + cooldown: defaultCooldown, + reconnectBackoff: defaultReconnectBackoff, + logger: logger.With().Str("component", "verification_monitor").Logger(), } } @@ -91,7 +94,9 @@ type exporter struct { workers int retryIntervals []time.Duration cooldown time.Duration - logger zerolog.Logger + // reconnectBackoff is the initial delay before re-subscribing after the subscription drops. + reconnectBackoff time.Duration + logger zerolog.Logger } // ExportMetrics starts the block verification monitoring loop @@ -101,6 +106,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error if err != nil { return err } + m.SetSubscriptionUp(true) // create buffered channel for block queue blockQueue := make(chan job, e.workers*2) @@ -123,6 +129,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error m.InitializeSubmissionMetrics(e.chainID) m.InitializeBlockTimeSloThresholds(e.chainID) m.InitializeBlockReceiveDelaySloThresholds(e.chainID) + m.InitializeVerifierMetrics(e.chainID) // ticker to refresh submission duration metric every 10 seconds refreshTicker := time.NewTicker(10 * time.Second) @@ -153,6 +160,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error e.logger.Warn().Msg("WebSocket subscription closed, reconnecting") } sub.Unsubscribe() + m.SetSubscriptionUp(false) newSub := e.reconnectSubscription(ctx, headers) if newSub == nil { // context was cancelled during reconnection @@ -160,16 +168,20 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error return nil } sub = newSub + m.SetSubscriptionUp(true) + m.RecordWsReconnect(e.chainID) e.logger.Info().Msg("WebSocket subscription re-established") case <-refreshTicker.C: // ensure that submission duration is always included in the 60 second window. m.RefreshSubmissionDuration() // update time since last block metric m.UpdateTimeSinceLastBlock() + m.SetVerifierQueueLength(e.chainID, len(blockQueue)) case header := <-headers: // record block arrival time for millisecond precision arrivalTime := time.Now() m.RecordBlockTime(e.chainID, arrivalTime) + m.RecordHeaderReceived(e.chainID, arrivalTime) m.RecordBlockReceiveDelay(e.chainID, max(0, arrivalTime.Sub(time.Unix(int64(header.Time), 0)))) height := header.Number.Uint64() @@ -182,7 +194,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error workerGroup.Add(1) go func(from, to uint64) { defer workerGroup.Done() - e.backfill(ctx, blockQueue, from, to) + e.backfill(ctx, m, blockQueue, from, to) }(lastHeight+1, height-1) } lastHeight = max(lastHeight, height) @@ -202,7 +214,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error // reconnectSubscription attempts to re-establish the WebSocket block header subscription // with exponential backoff. Returns nil if the context is cancelled before reconnecting. func (e *exporter) reconnectSubscription(ctx context.Context, headers chan *types.Header) ethereum.Subscription { - backoff := 5 * time.Second + backoff := e.reconnectBackoff const maxBackoff = 60 * time.Second for { @@ -228,8 +240,9 @@ func (e *exporter) reconnectSubscription(ctx context.Context, headers chan *type // backfill enqueues the blocks in [from, to] that were missed by the subscription, // keeping only the most recent maxBackfill heights. -func (e *exporter) backfill(ctx context.Context, blockQueue chan<- job, from, to uint64) { +func (e *exporter) backfill(ctx context.Context, m *metrics.Metrics, blockQueue chan<- job, from, to uint64) { if n := to - from + 1; n > maxBackfill { + m.RecordBackfillDropped(e.chainID, n-maxBackfill) e.logger.Warn().Uint64("dropped", n-maxBackfill).Uint64("from", from).Uint64("to", to).Msg("backfill gap too large, skipping oldest blocks") from = to - maxBackfill + 1 } @@ -246,6 +259,7 @@ func (e *exporter) backfill(ctx context.Context, blockQueue chan<- job, from, to } select { case blockQueue <- job{header: header}: + m.RecordBackfill(e.chainID, 1) case <-ctx.Done(): return } @@ -317,6 +331,7 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue logger := e.logger.With().Str("namespace", namespace).Uint64("block_height", blockHeight).Logger() if j.start.IsZero() { j.start = time.Now() + m.AddInflightBlocks(e.chainID, 1) logger.Debug(). Str("hash", header.Hash().Hex()). Time("time", blockTime). @@ -326,6 +341,7 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue } if e.verifyAttempt(ctx, m, logger, j.attempt+1, blockHeight, namespace, blockTime, j.start) { + m.AddInflightBlocks(e.chainID, -1) return } @@ -337,6 +353,7 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue logger.Error().Msg("max retries exhausted: failed to verify block") e.onVerified(m, namespace, blockHeight, 0, false, 0) + m.AddInflightBlocks(e.chainID, -1) enqueueAfter(ctx, blockQueue, e.cooldown, job{header: header}) } diff --git a/pkg/exporters/verifier/verifier_test.go b/pkg/exporters/verifier/verifier_test.go index 5008cc5..5715c2d 100644 --- a/pkg/exporters/verifier/verifier_test.go +++ b/pkg/exporters/verifier/verifier_test.go @@ -2,6 +2,7 @@ package verifier import ( "context" + "errors" "math/big" "slices" "sync" @@ -76,6 +77,7 @@ func (s fakeSub) Err() <-chan error { return s.err } type fakeHeads struct { mu sync.Mutex ch chan<- *types.Header + subs []fakeSub fetched []uint64 } @@ -83,7 +85,9 @@ func (f *fakeHeads) SubscribeNewHead(_ context.Context, ch chan<- *types.Header) f.mu.Lock() defer f.mu.Unlock() f.ch = ch - return fakeSub{err: make(chan error)}, nil + sub := fakeSub{err: make(chan error, 1)} + f.subs = append(f.subs, sub) + return sub, nil } func (f *fakeHeads) HeaderByNumber(_ context.Context, number *big.Int) (*types.Header, error) { @@ -93,6 +97,12 @@ func (f *fakeHeads) HeaderByNumber(_ context.Context, number *big.Int) (*types.H return newHeader(number.Uint64()), nil } +func (f *fakeHeads) subscriptions() []fakeSub { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.subs) +} + func (f *fakeHeads) fetchedHeights() []uint64 { f.mu.Lock() defer f.mu.Unlock() @@ -123,6 +133,7 @@ func newHarness(workers int, notYet map[uint64]int, retryIntervals ...time.Durat h.e = NewMetricsExporter(h.blocks, h.da, h.heads, []byte("h"), []byte("d"), "chain", workers, zerolog.Nop()).(*exporter) h.e.retryIntervals = retryIntervals h.e.cooldown = time.Hour + h.e.reconnectBackoff = time.Millisecond return h } @@ -341,15 +352,104 @@ func TestExportMetricsRecordsReceiveDelayAndSloGauges(t *testing.T) { } } +func TestExportMetricsReconnectUpdatesCounterAndHealth(t *testing.T) { + h := newHarness(1, nil, 0, time.Millisecond) + if ok, _ := h.m.CheckHealth(0); ok { + t.Fatal("expected not ready before the subscription is established") + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error) + go func() { done <- h.e.ExportMetrics(ctx, h.m) }() + defer func() { + cancel() + if err := <-done; err != nil { + t.Error(err) + } + }() + + eventually(t, "not subscribed", func() bool { return len(h.heads.subscriptions()) == 1 }) + if ok, reason := h.m.CheckHealth(0); !ok { + t.Fatalf("expected healthy after subscribing, got %q", reason) + } + if got := h.metric(t, "ws_reconnects_total"); got != 0 { + t.Fatalf("expected no reconnects yet, got %v", got) + } + + h.heads.subs[0].err <- errors.New("dropped") + + eventually(t, "reconnect not counted", func() bool { return h.metric(t, "ws_reconnects_total") == 1 }) + eventually(t, "health not restored", func() bool { + ok, _ := h.m.CheckHealth(0) + return ok + }) + if got := len(h.heads.subscriptions()); got != 2 { + t.Fatalf("expected 2 subscriptions, got %d", got) + } +} + +func TestExportMetricsRecordsLastHeaderAndInflightGauges(t *testing.T) { + h := newHarness(1, map[uint64]int{1: -1}, 0, time.Hour) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error) + go func() { done <- h.e.ExportMetrics(ctx, h.m) }() + defer func() { + cancel() + if err := <-done; err != nil { + t.Error(err) + } + }() + + if got := h.metric(t, "last_header_received_timestamp"); got != 0 { + t.Fatalf("expected zero timestamp before any header, got %v", got) + } + eventually(t, "not subscribed", func() bool { return len(h.heads.subscriptions()) == 1 }) + h.heads.mu.Lock() + ch := h.heads.ch + h.heads.mu.Unlock() + ch <- newHeader(1) + + eventually(t, "last header not recorded", func() bool { return h.metric(t, "last_header_received_timestamp") > 0 }) + eventually(t, "block not in flight", func() bool { return h.metric(t, "verifier_inflight_blocks") == 1 }) + if ok, reason := h.m.CheckHealth(time.Minute); !ok { + t.Fatalf("expected healthy with a fresh header, got %q", reason) + } +} + +func TestInflightGaugeReturnsToZeroOnceVerified(t *testing.T) { + h := newHarness(1, map[uint64]int{1: 2}, 0, time.Millisecond, time.Millisecond) + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "block not verified", func() bool { return len(h.da.verifiedHeights()) == 1 }) + eventually(t, "inflight not decremented", func() bool { return h.metric(t, "verifier_inflight_blocks") == 0 }) +} + +func TestInflightGaugeReturnsToZeroOnExhaustion(t *testing.T) { + h := newHarness(1, map[uint64]int{1: -1}, 0, time.Millisecond) + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "missing block not recorded", func() bool { return h.metric(t, "unsubmitted_blocks_total") == 1 }) + eventually(t, "inflight not decremented", func() bool { return h.metric(t, "verifier_inflight_blocks") == 0 }) +} + func TestBackfillCapsGap(t *testing.T) { h := newHarness(1, nil) queue := make(chan job, maxBackfill+10) - h.e.backfill(context.Background(), queue, 1, maxBackfill+5) + h.e.backfill(context.Background(), h.m, queue, 1, maxBackfill+5) if got := len(queue); got != maxBackfill { t.Fatalf("expected %d backfilled jobs, got %d", maxBackfill, got) } + if got := h.metric(t, "verifier_backfilled_blocks_total"); got != maxBackfill { + t.Fatalf("expected %d backfilled blocks counted, got %v", maxBackfill, got) + } + if got := h.metric(t, "verifier_backfill_dropped_total"); got != 5 { + t.Fatalf("expected 5 dropped blocks counted, got %v", got) + } if first := (<-queue).header.Number.Uint64(); first != 6 { t.Fatalf("expected backfill to keep the most recent heights starting at 6, got %d", first) } diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index be78992..0b6952b 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -7,6 +7,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/prometheus/client_golang/prometheus" @@ -68,6 +69,25 @@ type Metrics struct { // ConsensusRpcEndpointErrors tracks consensus RPC endpoint errors by type. ConsensusRpcEndpointErrors *prometheus.CounterVec + // VerifierQueueLength tracks the number of blocks waiting in the verifier queue. + VerifierQueueLength *prometheus.GaugeVec + // VerifierInflightBlocks tracks blocks between their first verification attempt and being verified or exhausted. + VerifierInflightBlocks *prometheus.GaugeVec + // WsReconnectsTotal tracks successful re-subscriptions of the block header WebSocket. + WsReconnectsTotal *prometheus.CounterVec + // LastHeaderReceivedTimestamp tracks the unix time at which the last header arrived from the subscription. + LastHeaderReceivedTimestamp *prometheus.GaugeVec + // VerifierBackfilledBlocksTotal tracks blocks enqueued to cover gaps in the header subscription. + VerifierBackfilledBlocksTotal *prometheus.CounterVec + // VerifierBackfillDroppedTotal tracks blocks skipped because a gap exceeded the backfill cap. + VerifierBackfillDroppedTotal *prometheus.CounterVec + + // health state read by the /health endpoint + subscriptionUp atomic.Bool + subscriptionSeen atomic.Bool + subscriptionSince atomic.Int64 // unix nanos of the first established subscription + lastHeaderReceived atomic.Int64 // unix nanos + // internal tracking to ensure we only record increasing DA heights latestHeaderDaHeight uint64 latestDataDaHeight uint64 @@ -336,6 +356,54 @@ func NewWithRegistry(namespace string, registerer prometheus.Registerer) *Metric }, []string{"chain_id", "type"}, ), + VerifierQueueLength: factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "verifier_queue_length", + Help: "number of blocks waiting in the verifier queue", + }, + []string{"chain_id"}, + ), + VerifierInflightBlocks: factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "verifier_inflight_blocks", + Help: "number of blocks being verified, including those waiting for a retry", + }, + []string{"chain_id"}, + ), + WsReconnectsTotal: factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "ws_reconnects_total", + Help: "total number of successful block header WebSocket re-subscriptions", + }, + []string{"chain_id"}, + ), + LastHeaderReceivedTimestamp: factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "last_header_received_timestamp", + Help: "unix timestamp in seconds of the last block header received from the subscription", + }, + []string{"chain_id"}, + ), + VerifierBackfilledBlocksTotal: factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "verifier_backfilled_blocks_total", + Help: "total number of blocks backfilled after gaps in the header subscription", + }, + []string{"chain_id"}, + ), + VerifierBackfillDroppedTotal: factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "verifier_backfill_dropped_total", + Help: "total number of blocks skipped because a subscription gap exceeded the backfill cap", + }, + []string{"chain_id"}, + ), ranges: make(map[string][]*blockRange), lastBlockArrivalTime: make(map[string]time.Time), lastSubmissionDurations: make(map[string]time.Duration), @@ -720,6 +788,77 @@ func (m *Metrics) InitializeSubmissionMetrics(chainID string) { } } +// InitializeVerifierMetrics pre-initializes the verifier self-observability metrics +// so they are visible from startup. The last header timestamp stays 0 until the +// first header arrives. +func (m *Metrics) InitializeVerifierMetrics(chainID string) { + m.VerifierQueueLength.WithLabelValues(chainID).Set(0) + m.VerifierInflightBlocks.WithLabelValues(chainID).Set(0) + m.WsReconnectsTotal.WithLabelValues(chainID).Add(0) + m.LastHeaderReceivedTimestamp.WithLabelValues(chainID).Set(0) + m.VerifierBackfilledBlocksTotal.WithLabelValues(chainID).Add(0) + m.VerifierBackfillDroppedTotal.WithLabelValues(chainID).Add(0) +} + +// SetVerifierQueueLength records the current verifier queue length. +func (m *Metrics) SetVerifierQueueLength(chainID string, n int) { + m.VerifierQueueLength.WithLabelValues(chainID).Set(float64(n)) +} + +// AddInflightBlocks adjusts the in-flight verification gauge by delta. +func (m *Metrics) AddInflightBlocks(chainID string, delta float64) { + m.VerifierInflightBlocks.WithLabelValues(chainID).Add(delta) +} + +// RecordWsReconnect counts a re-established header subscription. +func (m *Metrics) RecordWsReconnect(chainID string) { + m.WsReconnectsTotal.WithLabelValues(chainID).Inc() +} + +// RecordBackfill counts blocks enqueued by a backfill. +func (m *Metrics) RecordBackfill(chainID string, enqueued uint64) { + m.VerifierBackfilledBlocksTotal.WithLabelValues(chainID).Add(float64(enqueued)) +} + +// RecordBackfillDropped counts blocks skipped because a gap exceeded the backfill cap. +func (m *Metrics) RecordBackfillDropped(chainID string, dropped uint64) { + m.VerifierBackfillDroppedTotal.WithLabelValues(chainID).Add(float64(dropped)) +} + +// SetSubscriptionUp records whether the block header subscription is currently established. +func (m *Metrics) SetSubscriptionUp(up bool) { + if up && m.subscriptionSeen.CompareAndSwap(false, true) { + m.subscriptionSince.Store(time.Now().UnixNano()) + } + m.subscriptionUp.Store(up) +} + +// RecordHeaderReceived records the arrival of a block header from the subscription. +func (m *Metrics) RecordHeaderReceived(chainID string, t time.Time) { + m.lastHeaderReceived.Store(t.UnixNano()) + m.LastHeaderReceivedTimestamp.WithLabelValues(chainID).Set(float64(t.Unix())) +} + +// CheckHealth reports whether the monitor is ready. It fails until a subscription +// has been established, while the subscription is down, and, when maxHeaderAge is +// positive, when no header has arrived for longer than maxHeaderAge (measured from +// the first subscription if no header has arrived yet). +func (m *Metrics) CheckHealth(maxHeaderAge time.Duration) (bool, string) { + if !m.subscriptionSeen.Load() { + return false, "not ready: block header subscription not established" + } + if !m.subscriptionUp.Load() { + return false, "block header subscription is down" + } + if maxHeaderAge > 0 { + last := max(m.lastHeaderReceived.Load(), m.subscriptionSince.Load()) + if age := time.Since(time.Unix(0, last)); age > maxHeaderAge { + return false, fmt.Sprintf("no block header received for %s (limit %s)", age.Round(time.Second), maxHeaderAge) + } + } + return true, "" +} + // InitializeJsonRpcSloThresholds initializes the constant SLO threshold gauges for JSON-RPC requests func (m *Metrics) InitializeJsonRpcSloThresholds(chainID string) { m.JsonRpcRequestSloSeconds.WithLabelValues(chainID, "0.5").Set(0.2) diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index 70afc0f..6cb227c 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -751,3 +751,81 @@ func getMetricValue(t *testing.T, reg *prometheus.Registry, metricName string, l t.Fatalf("metric %s not found with labels %v", metricName, labels) return 0 } + +func gatheredValue(t *testing.T, reg *prometheus.Registry, name string) (float64, bool) { + t.Helper() + families, err := reg.Gather() + require.NoError(t, err) + for _, mf := range families { + if mf.GetName() != name { + continue + } + require.Len(t, mf.GetMetric(), 1) + m := mf.GetMetric()[0] + if m.Gauge != nil { + return m.GetGauge().GetValue(), true + } + return m.GetCounter().GetValue(), true + } + return 0, false +} + +func TestMetrics_VerifierSelfObservability(t *testing.T) { + reg := prometheus.NewRegistry() + m := NewWithRegistry("test", reg) + names := []string{ + "test_verifier_queue_length", + "test_verifier_inflight_blocks", + "test_ws_reconnects_total", + "test_last_header_received_timestamp", + "test_verifier_backfilled_blocks_total", + "test_verifier_backfill_dropped_total", + } + + m.InitializeVerifierMetrics("chain") + for _, name := range names { + v, ok := gatheredValue(t, reg, name) + require.True(t, ok, name) + require.Zero(t, v, name) + } + + m.SetVerifierQueueLength("chain", 7) + m.AddInflightBlocks("chain", 1) + m.AddInflightBlocks("chain", 1) + m.AddInflightBlocks("chain", -1) + m.RecordWsReconnect("chain") + m.RecordWsReconnect("chain") + m.RecordHeaderReceived("chain", time.Unix(1700000000, 0)) + m.RecordBackfill("chain", 3) + m.RecordBackfillDropped("chain", 5) + + for name, want := range map[string]float64{ + "test_verifier_queue_length": 7, + "test_verifier_inflight_blocks": 1, + "test_ws_reconnects_total": 2, + "test_last_header_received_timestamp": 1700000000, + "test_verifier_backfilled_blocks_total": 3, + "test_verifier_backfill_dropped_total": 5, + } { + v, ok := gatheredValue(t, reg, name) + require.True(t, ok, name) + require.Equal(t, want, v, name) + } +} + +func TestMetrics_CheckHealth(t *testing.T) { + m := NewWithRegistry("test", prometheus.NewRegistry()) + + ok, reason := m.CheckHealth(0) + require.False(t, ok) + require.Contains(t, reason, "not ready") + + m.SetSubscriptionUp(true) + ok, _ = m.CheckHealth(0) + require.True(t, ok) + + m.SetSubscriptionUp(false) + ok, reason = m.CheckHealth(0) + require.False(t, ok) + require.Contains(t, reason, "down") +} diff --git a/pkg/metrics/server.go b/pkg/metrics/server.go index e023bf3..5079991 100644 --- a/pkg/metrics/server.go +++ b/pkg/metrics/server.go @@ -18,21 +18,20 @@ type Exporter interface { } // StartServer starts an HTTP server to expose Prometheus metrics and a health check endpoint. +// /health is a readiness check: it fails until the header subscription is up, while it is +// down, and (when healthMaxHeaderAge > 0) when no header has arrived within that duration. // The server shuts down gracefully when the context is cancelled. -func StartServer(ctx context.Context, metricsPath string, port int, logger zerolog.Logger, exporters ...Exporter) error { +func StartServer(ctx context.Context, metricsPath string, port int, healthMaxHeaderAge time.Duration, logger zerolog.Logger, exporters ...Exporter) error { // create logger with component field logger = logger.With().Str("component", "http_server").Logger() - mux := http.NewServeMux() - mux.Handle(metricsPath, promhttp.Handler()) - mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte("OK")) - }) - // initialize metrics (thread-safe, shared across all exporters) m := New("ev_metrics") + mux := http.NewServeMux() + mux.Handle(metricsPath, promhttp.Handler()) + mux.HandleFunc("/health", healthHandler(m, healthMaxHeaderAge)) + // derived ctx is cancelled when any exporter or the server fails, stopping everything else g, ctx := errgroup.WithContext(ctx) @@ -85,3 +84,14 @@ func StartServer(ctx context.Context, metricsPath string, port int, logger zerol return g.Wait() } + +func healthHandler(m *Metrics, maxHeaderAge time.Duration) http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { + ok, reason := m.CheckHealth(maxHeaderAge) + if !ok { + http.Error(w, reason, http.StatusServiceUnavailable) + return + } + _, _ = w.Write([]byte("OK")) + } +} diff --git a/pkg/metrics/server_test.go b/pkg/metrics/server_test.go index 8413000..0ee68d4 100644 --- a/pkg/metrics/server_test.go +++ b/pkg/metrics/server_test.go @@ -4,9 +4,12 @@ import ( "context" "errors" "net" + "net/http" + "net/http/httptest" "testing" "time" + "github.com/prometheus/client_golang/prometheus" "github.com/rs/zerolog" "github.com/stretchr/testify/require" ) @@ -24,7 +27,7 @@ func TestStartServer_ExporterErrorStopsServer(t *testing.T) { wantErr := errors.New("subscribe failed") done := make(chan error, 1) go func() { - done <- StartServer(context.Background(), "/metrics", port, zerolog.Nop(), failingExporter{err: wantErr}) + done <- StartServer(context.Background(), "/metrics", port, 0, zerolog.Nop(), failingExporter{err: wantErr}) }() select { @@ -34,3 +37,62 @@ func TestStartServer_ExporterErrorStopsServer(t *testing.T) { t.Fatal("StartServer did not return after exporter failure") } } + +func healthStatus(t *testing.T, m *Metrics, maxAge time.Duration) (int, string) { + t.Helper() + rec := httptest.NewRecorder() + healthHandler(m, maxAge)(rec, httptest.NewRequest(http.MethodGet, "/health", nil)) + return rec.Code, rec.Body.String() +} + +func TestHealthHandler(t *testing.T) { + m := NewWithRegistry("test", prometheus.NewRegistry()) + + code, body := healthStatus(t, m, 0) + require.Equal(t, http.StatusServiceUnavailable, code) + require.Contains(t, body, "not ready") + + m.SetSubscriptionUp(true) + code, body = healthStatus(t, m, 0) + require.Equal(t, http.StatusOK, code) + require.Contains(t, body, "OK") + + m.SetSubscriptionUp(false) + code, body = healthStatus(t, m, 0) + require.Equal(t, http.StatusServiceUnavailable, code) + require.Contains(t, body, "down") + + m.SetSubscriptionUp(true) + code, _ = healthStatus(t, m, 0) + require.Equal(t, http.StatusOK, code) +} + +func TestHealthHandlerHeaderAge(t *testing.T) { + m := NewWithRegistry("test", prometheus.NewRegistry()) + m.SetSubscriptionUp(true) + m.RecordHeaderReceived("chain", time.Now()) + time.Sleep(20 * time.Millisecond) + + // a stale header fails only when the age check is enabled + code, _ := healthStatus(t, m, 0) + require.Equal(t, http.StatusOK, code) + code, body := healthStatus(t, m, time.Millisecond) + require.Equal(t, http.StatusServiceUnavailable, code) + require.Contains(t, body, "no block header received") + + m.RecordHeaderReceived("chain", time.Now()) + code, _ = healthStatus(t, m, time.Minute) + require.Equal(t, http.StatusOK, code) +} + +func TestHealthHandlerHeaderAgeBeforeFirstHeader(t *testing.T) { + m := NewWithRegistry("test", prometheus.NewRegistry()) + m.SetSubscriptionUp(true) + + code, _ := healthStatus(t, m, time.Minute) + require.Equal(t, http.StatusOK, code) + + time.Sleep(20 * time.Millisecond) + code, _ = healthStatus(t, m, time.Millisecond) + require.Equal(t, http.StatusServiceUnavailable, code) +} From 89489f26b7d0494329dee7b29c748ef9a9cae301 Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:38:37 +0200 Subject: [PATCH 5/7] feat(verifier): verify header blobs for blocks with transactions too Blocks with transactions now verify both the header and data blobs, each recorded under its own type label. Per-job progress means retries only redo the pending part, and exhaustion records only unverified types as missing. Fixes #47 Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 + pkg/exporters/verifier/verifier.go | 131 +++++++++---------- pkg/exporters/verifier/verifier_test.go | 161 ++++++++++++++++++++++-- 3 files changed, 215 insertions(+), 79 deletions(-) diff --git a/README.md b/README.md index 67a7a50..ffd69e3 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,8 @@ This will query account balances every 30 seconds with automatic failover betwee The following metrics are exposed: +Blocks without transactions have only their header blob verified (`type="header"`). Blocks with transactions have both their header blob (`type="header"`) and data blob (`type="data"`) verified, and each outcome is recorded under its own type. Tx blocks now count under `type="header"` too, so header submission counts (`submission_*`, `unsubmitted_*`, DA height and duration) rise after upgrading. Each block costs one extra Celestia request per transaction block for the header blob. If retries are exhausted, only the blob types that are still unverified are recorded as missing. + ### `ev_metrics_unsubmitted_block_range_start` - **Type**: Gauge - **Labels**: `chain_id`, `blob_type`, `range_id` diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 556e8cd..6284bd3 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -31,11 +31,14 @@ type headSource interface { } // job is one pending verification attempt for a block. start is zero until the -// block is first processed by a worker. +// block is first processed by a worker. headerDone and dataDone record which blobs +// are already verified so retries do not repeat or double count them. type job struct { - header *types.Header - attempt int - start time.Time + header *types.Header + attempt int + start time.Time + headerDone bool + dataDone bool } // defaultRetryIntervals is the delay before each verification attempt, matching @@ -314,6 +317,9 @@ const verifyAttemptTimeout = 30 * time.Second // verifyBlock makes one verification attempt for j and schedules the next attempt // (or a cooldown re-queue once retries are exhausted) without blocking the worker. +// Blocks with transactions verify both their header and data blobs; blocks without +// only verify the header. Each part is tracked on the job so retries only redo what +// is still pending. func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQueue chan<- job, j job) { header := j.header blockHeight := header.Number.Uint64() @@ -321,14 +327,9 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue // check if block has transactions hasTransactions := header.TxHash != types.EmptyRootHash - namespace := "header" - if hasTransactions { - namespace = "data" - } - blockTime := time.Unix(int64(header.Time), 0) - logger := e.logger.With().Str("namespace", namespace).Uint64("block_height", blockHeight).Logger() + logger := e.logger.With().Uint64("block_height", blockHeight).Logger() if j.start.IsZero() { j.start = time.Now() m.AddInflightBlocks(e.chainID, 1) @@ -340,7 +341,7 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue Msg("processing block") } - if e.verifyAttempt(ctx, m, logger, j.attempt+1, blockHeight, namespace, blockTime, j.start) { + if e.verifyAttempt(ctx, m, logger, &j, hasTransactions, blockTime) { m.AddInflightBlocks(e.chainID, -1) return } @@ -352,91 +353,81 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQue } logger.Error().Msg("max retries exhausted: failed to verify block") - e.onVerified(m, namespace, blockHeight, 0, false, 0) + if !j.headerDone { + e.onVerified(m, "header", blockHeight, 0, false, 0) + } + if hasTransactions && !j.dataDone { + e.onVerified(m, "data", blockHeight, 0, false, 0) + } m.AddInflightBlocks(e.chainID, -1) enqueueAfter(ctx, blockQueue, e.cooldown, job{header: header}) } -// verifyAttempt performs one bounded RPC attempt to verify a block against Celestia DA. -// It returns true when retrying is no longer needed (verified, or permanent failure), -// and false when the caller should retry. -// Each call is bounded by verifyAttemptTimeout so workers cannot hang indefinitely +// verifyAttempt performs one bounded RPC attempt to verify the pending parts of a block +// against Celestia DA, updating j.headerDone and j.dataDone. It returns true once every +// required part is verified and false when the caller should retry. +// The attempt is bounded by verifyAttemptTimeout so workers cannot hang indefinitely // on slow or unresponsive ev-node / Celestia endpoints. -func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger zerolog.Logger, retries int, blockHeight uint64, namespace string, blockTime time.Time, startTime time.Time) bool { +func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger zerolog.Logger, j *job, hasTransactions bool, blockTime time.Time) bool { attemptCtx, cancel := context.WithTimeout(ctx, verifyAttemptTimeout) defer cancel() + retries := j.attempt + 1 + blockHeight := j.header.Number.Uint64() + blockResultWithBlobs, err := e.evnodeClient.GetBlockWithBlobs(attemptCtx, blockHeight) if err != nil { logger.Warn().Err(err).Int("attempt", retries).Msg("failed to query block from ev-node") return false } - daHeight := blockResultWithBlobs.HeaderDaHeight - if namespace == "data" { - daHeight = blockResultWithBlobs.DataDaHeight - } - - if daHeight == 0 { - logger.Debug().Int("attempt", retries).Msg("block still not submitted to DA, will retry") - return false - } - - daBlockTime, err := e.celestiaClient.GetBlockTimestamp(attemptCtx, daHeight) - if err != nil { - logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("failed to get da block timestamp") - return false - } - - // the time taken from block time to DA inclusion time. - submissionDuration := daBlockTime.Sub(blockTime) + verifyPart := func(namespace string, daHeight uint64, verify func(daHeight uint64) (bool, error)) bool { + logger := logger.With().Str("namespace", namespace).Logger() + if daHeight == 0 { + logger.Debug().Int("attempt", retries).Msg("block still not submitted to DA, will retry") + return false + } - switch namespace { - case "header": - verified, err := e.celestiaClient.VerifyBlobAtHeight(attemptCtx, blockResultWithBlobs.HeaderBlob, daHeight, e.headerNS) + daBlockTime, err := e.celestiaClient.GetBlockTimestamp(attemptCtx, daHeight) if err != nil { - logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("verification failed") + logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("failed to get da block timestamp") return false } - if verified { - logger.Debug(). - Uint64("da_height", daHeight). - Dur("duration", time.Since(startTime)). - Msg("header blob verified on Celestia") - e.onVerified(m, namespace, blockHeight, daHeight, true, submissionDuration) - return true - } - logger.Warn().Uint64("da_height", daHeight).Int("attempt", retries).Msg("header verification failed, will retry") - - case "data": - if len(blockResultWithBlobs.DataBlob) == 0 { - logger.Debug(). - Dur("duration", time.Since(startTime)). - Msg("empty data block - no verification needed") - e.onVerified(m, namespace, blockHeight, daHeight, true, submissionDuration) - return true - } - // perform actual verification between bytes from ev-node and Celestia. - verified, err := e.celestiaClient.VerifyDataBlobAtHeight(attemptCtx, blockResultWithBlobs.DataBlob, daHeight, e.dataNS) + verified, err := verify(daHeight) if err != nil { logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("verification failed") return false } - if verified { - logger.Debug(). - Uint64("da_height", daHeight). - Dur("duration", time.Since(startTime)). - Msg("data blob verified on Celestia") - e.onVerified(m, namespace, blockHeight, daHeight, true, submissionDuration) - return true + if !verified { + logger.Warn().Uint64("da_height", daHeight).Int("attempt", retries).Msg("verification failed, will retry") + return false } - logger.Warn().Uint64("da_height", daHeight).Int("attempt", retries).Msg("verification failed, will retry") - default: - logger.Error().Str("namespace", namespace).Msg("unknown namespace type") + logger.Debug(). + Uint64("da_height", daHeight). + Dur("duration", time.Since(j.start)). + Msg("blob verified on Celestia") + // the time taken from block time to DA inclusion time. + e.onVerified(m, namespace, blockHeight, daHeight, true, daBlockTime.Sub(blockTime)) return true } - return false + if !j.headerDone { + j.headerDone = verifyPart("header", blockResultWithBlobs.HeaderDaHeight, func(daHeight uint64) (bool, error) { + return e.celestiaClient.VerifyBlobAtHeight(attemptCtx, blockResultWithBlobs.HeaderBlob, daHeight, e.headerNS) + }) + } + + if hasTransactions && !j.dataDone { + j.dataDone = verifyPart("data", blockResultWithBlobs.DataDaHeight, func(daHeight uint64) (bool, error) { + if len(blockResultWithBlobs.DataBlob) == 0 { + logger.Debug().Msg("empty data block - no verification needed") + return true, nil + } + return e.celestiaClient.VerifyDataBlobAtHeight(attemptCtx, blockResultWithBlobs.DataBlob, daHeight, e.dataNS) + }) + } + + return j.headerDone && (!hasTransactions || j.dataDone) } diff --git a/pkg/exporters/verifier/verifier_test.go b/pkg/exporters/verifier/verifier_test.go index 5715c2d..98ffb58 100644 --- a/pkg/exporters/verifier/verifier_test.go +++ b/pkg/exporters/verifier/verifier_test.go @@ -10,6 +10,7 @@ import ( "time" ethereum "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/evstack/ev-metrics/internal/clients/evnode" "github.com/evstack/ev-metrics/pkg/metrics" @@ -19,22 +20,28 @@ import ( const daOffset = 100 -// fakeBlocks reports a block as not yet submitted for notYet[height] attempts -// (forever if negative), then as submitted at DA height height+daOffset. +// fakeBlocks reports a block's header as not yet submitted for notYet[height] attempts +// (forever if negative), then as submitted at DA height height+daOffset. The data blob +// behaves the same way using dataNotYet. type fakeBlocks struct { - mu sync.Mutex - notYet map[uint64]int - calls map[uint64]int + mu sync.Mutex + notYet map[uint64]int + dataNotYet map[uint64]int + calls map[uint64]int } func (f *fakeBlocks) GetBlockWithBlobs(_ context.Context, height uint64) (*evnode.GetBlockResult, error) { f.mu.Lock() defer f.mu.Unlock() f.calls[height]++ - if n := f.notYet[height]; n < 0 || f.calls[height] <= n { - return &evnode.GetBlockResult{}, nil + res := &evnode.GetBlockResult{} + if n := f.notYet[height]; n >= 0 && f.calls[height] > n { + res.HeaderDaHeight, res.HeaderBlob = height+daOffset, []byte{1} } - return &evnode.GetBlockResult{HeaderDaHeight: height + daOffset, HeaderBlob: []byte{1}}, nil + if n := f.dataNotYet[height]; n >= 0 && f.calls[height] > n { + res.DataDaHeight, res.DataBlob = height+daOffset, []byte{2} + } + return res, nil } func (f *fakeBlocks) callCount(height uint64) int { @@ -46,6 +53,7 @@ func (f *fakeBlocks) callCount(height uint64) int { type fakeDA struct { mu sync.Mutex verified []uint64 + data []uint64 } func (f *fakeDA) GetBlockTimestamp(context.Context, uint64) (time.Time, error) { @@ -59,10 +67,19 @@ func (f *fakeDA) VerifyBlobAtHeight(_ context.Context, _ []byte, daHeight uint64 return true, nil } -func (f *fakeDA) VerifyDataBlobAtHeight(context.Context, []byte, uint64, []byte) (bool, error) { +func (f *fakeDA) VerifyDataBlobAtHeight(_ context.Context, _ []byte, daHeight uint64, _ []byte) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.data = append(f.data, daHeight-daOffset) return true, nil } +func (f *fakeDA) dataHeights() []uint64 { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.data) +} + func (f *fakeDA) verifiedHeights() []uint64 { f.mu.Lock() defer f.mu.Unlock() @@ -113,6 +130,12 @@ func newHeader(height uint64) *types.Header { return &types.Header{Number: new(big.Int).SetUint64(height), TxHash: types.EmptyRootHash} } +func newTxHeader(height uint64) *types.Header { + h := newHeader(height) + h.TxHash = common.Hash{1} + return h +} + type harness struct { e *exporter blocks *fakeBlocks @@ -202,6 +225,29 @@ func (h *harness) sum(t *testing.T, name string) float64 { return sum } +// labeled sums the series of a metric whose label equals value. +func (h *harness) labeled(t *testing.T, name, label, value string) float64 { + t.Helper() + families, err := h.reg.Gather() + if err != nil { + t.Fatal(err) + } + var sum float64 + for _, mf := range families { + if mf.GetName() != "test_"+name { + continue + } + for _, m := range mf.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == label && l.GetValue() == value { + sum += m.GetGauge().GetValue() + m.GetCounter().GetValue() + } + } + } + } + return sum +} + func eventually(t *testing.T, msg string, cond func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) @@ -273,6 +319,103 @@ func TestRetriesExhaustedRequeuesAfterCooldown(t *testing.T) { } } +func TestTxBlockVerifiesHeaderAndData(t *testing.T) { + h := newHarness(1, nil, 0, time.Millisecond) + queue, _ := h.startWorkers(t) + + queue <- job{header: newTxHeader(1)} + + eventually(t, "block not verified", func() bool { return h.metric(t, "verifier_inflight_blocks") == 0 && len(h.da.dataHeights()) == 1 }) + if got := h.blocks.callCount(1); got != 1 { + t.Fatalf("expected 1 ev-node call, got %d", got) + } + for _, typ := range []string{"header", "data"} { + if got := h.labeled(t, "submission_attempts_total", "type", typ); got != 1 { + t.Fatalf("expected 1 %s attempt, got %v", typ, got) + } + if got := h.labeled(t, "submission_failures_total", "type", typ); got != 0 { + t.Fatalf("expected no %s failures, got %v", typ, got) + } + } + if got := h.da.verifiedHeights(); !slices.Equal(got, []uint64{1}) { + t.Fatalf("expected header verified once, got %v", got) + } +} + +func TestTxBlockHeaderVerifiedBeforeData(t *testing.T) { + h := newHarness(1, map[uint64]int{1: 1}, 0, time.Millisecond, time.Millisecond, time.Millisecond, time.Millisecond) + h.blocks.dataNotYet = map[uint64]int{1: 3} + queue, _ := h.startWorkers(t) + + queue <- job{header: newTxHeader(1)} + + eventually(t, "data not verified", func() bool { return len(h.da.dataHeights()) == 1 }) + eventually(t, "inflight not cleared", func() bool { return h.metric(t, "verifier_inflight_blocks") == 0 }) + if got := h.blocks.callCount(1); got != 4 { + t.Fatalf("expected 4 attempts, got %d", got) + } + if got := h.da.verifiedHeights(); !slices.Equal(got, []uint64{1}) { + t.Fatalf("expected header verified exactly once, got %v", got) + } + for _, typ := range []string{"header", "data"} { + if got := h.labeled(t, "submission_attempts_total", "type", typ); got != 1 { + t.Fatalf("expected 1 %s attempt, got %v", typ, got) + } + } + if got := h.metric(t, "unsubmitted_blocks_total"); got != 0 { + t.Fatalf("expected no unsubmitted blocks, got %v", got) + } +} + +func TestTxBlockMissingHeaderOnlyRecordsHeader(t *testing.T) { + h := newHarness(1, map[uint64]int{1: -1}, 0, time.Millisecond) + queue, stop := h.startWorkers(t) + + queue <- job{header: newTxHeader(1)} + + eventually(t, "missing block not recorded", func() bool { + return h.labeled(t, "unsubmitted_blocks_total", "blob_type", "header") == 1 + }) + stop() + if got := h.labeled(t, "unsubmitted_blocks_total", "blob_type", "data"); got != 0 { + t.Fatalf("expected data not to be missing, got %v", got) + } + if got := h.labeled(t, "unsubmitted_block_range_start", "blob_type", "data"); got != 0 { + t.Fatalf("expected no data range, got %v", got) + } + if got := h.labeled(t, "unsubmitted_block_range_start", "blob_type", "header"); got != 1 { + t.Fatalf("expected header range starting at 1, got %v", got) + } + if got := h.labeled(t, "submission_failures_total", "type", "header"); got != 1 { + t.Fatalf("expected 1 header failure, got %v", got) + } + if got := h.labeled(t, "submission_failures_total", "type", "data"); got != 0 { + t.Fatalf("expected no data failures, got %v", got) + } + if got := h.labeled(t, "submission_attempts_total", "type", "data"); got != 1 { + t.Fatalf("expected the verified data to count 1 attempt, got %v", got) + } + if got := h.da.dataHeights(); !slices.Equal(got, []uint64{1}) { + t.Fatalf("expected data verified once, got %v", got) + } +} + +func TestBlockWithoutTransactionsOnlyVerifiesHeader(t *testing.T) { + h := newHarness(1, nil, 0, time.Millisecond) + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "block not verified", func() bool { return len(h.da.verifiedHeights()) == 1 }) + eventually(t, "inflight not cleared", func() bool { return h.metric(t, "verifier_inflight_blocks") == 0 }) + if got := h.da.dataHeights(); len(got) != 0 { + t.Fatalf("expected no data verification, got %v", got) + } + if got := h.labeled(t, "submission_attempts_total", "type", "data"); got != 0 { + t.Fatalf("expected no data attempts, got %v", got) + } +} + func TestExportMetricsBackfillsGap(t *testing.T) { h := newHarness(2, nil, 0, time.Millisecond) ctx, cancel := context.WithCancel(context.Background()) From 98e20c2ee5517b5965e0af4814aa499de52e458c Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:39:26 +0200 Subject: [PATCH 6/7] docs: reword header-verification cost note Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 58c8771..890aa94 100644 --- a/README.md +++ b/README.md @@ -127,7 +127,7 @@ This will query account balances every 30 seconds with automatic failover betwee The following metrics are exposed: -Blocks without transactions have only their header blob verified (`type="header"`). Blocks with transactions have both their header blob (`type="header"`) and data blob (`type="data"`) verified, and each outcome is recorded under its own type. Tx blocks now count under `type="header"` too, so header submission counts (`submission_*`, `unsubmitted_*`, DA height and duration) rise after upgrading. Each block costs one extra Celestia request per transaction block for the header blob. If retries are exhausted, only the blob types that are still unverified are recorded as missing. +Blocks without transactions have only their header blob verified (`type="header"`). Blocks with transactions have both their header blob (`type="header"`) and data blob (`type="data"`) verified, and each outcome is recorded under its own type. Tx blocks now count under `type="header"` too, so header submission counts (`submission_*`, `unsubmitted_*`, DA height and duration) rise after upgrading. Each transaction block costs one extra Celestia request for its header blob. If retries are exhausted, only the blob types that are still unverified are recorded as missing. ### `ev_metrics_unsubmitted_block_range_start` - **Type**: Gauge From ee15ba2b40e95c1df26c1d3959a01666725ef0f9 Mon Sep 17 00:00:00 2001 From: auricom <27022259+auricom@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:42:41 +0200 Subject: [PATCH 7/7] fix(metrics): stop fabricating submission duration samples; add last_submission_duration_seconds RefreshSubmissionDuration re-observed the last value every 10s, which biased the summary quantiles and inflated _count and _sum. Remove it and expose the latest duration as a gauge instead. The summary now only contains real observations. Fixes #48 Co-Authored-By: Claude Sonnet 5.5 --- README.md | 7 ++++- pkg/exporters/verifier/verifier.go | 2 -- pkg/metrics/metrics.go | 41 ++++++++++-------------------- pkg/metrics/metrics_test.go | 32 +++++++++++++---------- 4 files changed, 39 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 890aa94..83ff0da 100644 --- a/README.md +++ b/README.md @@ -147,7 +147,12 @@ Blocks without transactions have only their header blob verified (`type="header" ### `ev_metrics_submission_duration_seconds` - **Type**: Summary - **Labels**: `chain_id`, `type` -- **Description**: DA blob submission duration from block creation to DA availability +- **Description**: DA blob submission duration from block creation to DA availability, over a 60-second rolling window. It only contains real observations, so the quantiles are `NaN` while no submission has been verified in the last 60 seconds. Use `ev_metrics_last_submission_duration_seconds` to see the latest value during quiet periods. + +### `ev_metrics_last_submission_duration_seconds` +- **Type**: Gauge +- **Labels**: `chain_id`, `type` +- **Description**: Duration from block creation to DA availability of the most recently verified submission. It keeps its value until the next submission is verified and is absent until the first one. ### `ev_metrics_submission_da_height` - **Type**: Gauge diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 6284bd3..69c98dd 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -175,8 +175,6 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error m.RecordWsReconnect(e.chainID) e.logger.Info().Msg("WebSocket subscription re-established") case <-refreshTicker.C: - // ensure that submission duration is always included in the 60 second window. - m.RefreshSubmissionDuration() // update time since last block metric m.UpdateTimeSinceLastBlock() m.SetVerifierQueueLength(e.chainID, len(blockQueue)) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 0b6952b..e62843a 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -29,7 +29,10 @@ type Metrics struct { // BlockHeightDrift tracks the drift between reference and target endpoints for a specific node. BlockHeightDrift *prometheus.GaugeVec // SubmissionDuration tracks DA blob submission duration percentiles over a rolling window. + // Only real observations are recorded, so quantiles are NaN while no submission is verified. SubmissionDuration *prometheus.SummaryVec + // LastSubmissionDuration tracks the duration of the most recently verified submission. + LastSubmissionDuration *prometheus.GaugeVec // SubmissionDaHeight tracks the DA height at which blocks were submitted. SubmissionDaHeight *prometheus.GaugeVec // SubmissionAttemptsTotal tracks the total number of submission attempts. @@ -95,9 +98,6 @@ type Metrics struct { // internal tracking for block time calculation (uses arrival time for ms precision) lastBlockArrivalTime map[string]time.Time // key: chainID - // lastSubmissionDurations tracks the most recent submission durations. - lastSubmissionDurations map[string]time.Duration // key: chainID:namespace - mu sync.Mutex ranges map[string][]*blockRange // key: blobType -> sorted slice of ranges } @@ -185,6 +185,14 @@ func NewWithRegistry(namespace string, registerer prometheus.Registerer) *Metric }, []string{"chain_id", "type"}, ), + LastSubmissionDuration: factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "last_submission_duration_seconds", + Help: "duration from block creation to da availability of the most recently verified submission", + }, + []string{"chain_id", "type"}, + ), SubmissionDaHeight: factory.NewGaugeVec( prometheus.GaugeOpts{ Namespace: namespace, @@ -404,9 +412,8 @@ func NewWithRegistry(namespace string, registerer prometheus.Registerer) *Metric }, []string{"chain_id"}, ), - ranges: make(map[string][]*blockRange), - lastBlockArrivalTime: make(map[string]time.Time), - lastSubmissionDurations: make(map[string]time.Duration), + ranges: make(map[string][]*blockRange), + lastBlockArrivalTime: make(map[string]time.Time), } return m @@ -684,28 +691,8 @@ func (m *Metrics) RecordBlockHeightDrift(chainID, targetEndpoint string, referen // RecordSubmissionDuration records the da submission duration for a given submission type func (m *Metrics) RecordSubmissionDuration(chainID, submissionType string, duration time.Duration) { - m.mu.Lock() - defer m.mu.Unlock() - m.SubmissionDuration.WithLabelValues(chainID, submissionType).Observe(duration.Seconds()) - - key := fmt.Sprintf("%s:%s", chainID, submissionType) - m.lastSubmissionDurations[key] = duration - -} - -// RefreshSubmissionDuration re-observes the last known submission duration to keep the metric alive. -func (m *Metrics) RefreshSubmissionDuration() { - m.mu.Lock() - defer m.mu.Unlock() - - for key, duration := range m.lastSubmissionDurations { - // assuming format "chainID:namespace" - parts := strings.Split(key, ":") - if len(parts) == 2 { - m.SubmissionDuration.WithLabelValues(parts[0], parts[1]).Observe(duration.Seconds()) - } - } + m.LastSubmissionDuration.WithLabelValues(chainID, submissionType).Set(duration.Seconds()) } // RecordBlockTime records the time between consecutive blocks using arrival time diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index 6cb227c..5d917e9 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -547,20 +547,26 @@ func TestMetrics_RecordSubmissionDuration(t *testing.T) { m.RecordSubmissionDuration("chain1", "data", 10*time.Second) m.RecordSubmissionDuration("chain2", "header", 3*time.Second) - // verify stored in memory - require.Equal(t, 6*time.Second, m.lastSubmissionDurations["chain1:header"], "last submission duration should be present") - require.Equal(t, 10*time.Second, m.lastSubmissionDurations["chain1:data"]) - require.Equal(t, 3*time.Second, m.lastSubmissionDurations["chain2:header"]) -} - -func TestMetrics_RefreshSubmissionDuration_Empty(t *testing.T) { - reg := prometheus.NewRegistry() - m := NewWithRegistry("test", reg) + // the gauge holds the latest value per chain and type + last := func(chain, typ string) float64 { + return getMetricValue(t, reg, "test_last_submission_duration_seconds", map[string]string{"chain_id": chain, "type": typ}) + } + require.Equal(t, 6.0, last("chain1", "header"), "last submission duration should be the most recent one") + require.Equal(t, 10.0, last("chain1", "data")) + require.Equal(t, 3.0, last("chain2", "header")) - // call refresh without any recorded values - should not panic - require.NotPanics(t, func() { - m.RefreshSubmissionDuration() - }) + // the summary only counts real observations: 4 calls above, none fabricated + families, err := reg.Gather() + require.NoError(t, err) + var count uint64 + for _, mf := range families { + if mf.GetName() == "test_submission_duration_seconds" { + for _, metric := range mf.GetMetric() { + count += metric.GetSummary().GetSampleCount() + } + } + } + require.EqualValues(t, 4, count) } func TestMetrics_RecordSubmissionAttempt(t *testing.T) {