P2: self-observability, /health, receive delay, tx-block header checks, URL reda

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`)
 
@@ -126,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 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
 - **Labels**: `chain_id`, `blob_type`, `range_id`
@@ -144,7 +147,12 @@ The following metrics are exposed:
 ### `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
@@ -194,7 +202,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,19 +212,60 @@ 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
   - `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:
@@ -241,6 +290,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 `#<index>` suffix to stay distinct.
+
 ### `ev_metrics_reference_block_height`
 - **Type**: Gauge
 - **Labels**: `chain_id`, `endpoint`

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"
@@ -45,6 +46,7 @@ const (
 	flagBalanceRpcUrls        = "balance.consensus-rpc-urls"
 	flagBalanceScrapeInterval = "balance.scrape-interval"
 	flagVerifierWorkers       = "verifier.workers"
+	flagHealthMaxHeaderAge    = "health-max-header-age"
 
 	metricsPath = "/metrics"
 )
@@ -72,6 +74,7 @@ type flagValues struct {
 	balanceRpcUrls        string
 	balanceScrapeInterval int
 	verifierWorkers       int
+	healthMaxHeaderAge    int
 }
 
 func NewMonitorCmd() *cobra.Command {
@@ -106,6 +109,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)
@@ -255,12 +259,12 @@ 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")
 	}
 
-	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) {

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")

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) {

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")
 		}
 

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)
+}