diff --git a/README.md b/README.md index f2ee322..51e9444 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Data Availability metrics and monitoring tool for evstack using Celestia DA. Thi - Automatic verification of header and data blobs on Celestia - Retry logic with exponential backoff for pending DA submissions - Prometheus metrics for tracking unverified block ranges -- Support for both streaming and one-shot block verification modes +- Streaming block verification of every new block - Account balance monitoring via Celestia consensus RPC with automatic failover ## Quick Start @@ -38,13 +38,14 @@ The monitor command streams EVM block headers and verifies DA submission on Cele ``` -### Enable Prometheus Metrics +### Prometheus Metrics Endpoint + +The Prometheus metrics server always runs. Use `--port` to change the port: ```bash ./ev-metrics monitor \ --header-namespace collect_testnet_header \ --data-namespace collect_testnet_data \ - --enable-metrics \ --port 2112 ``` @@ -66,7 +67,6 @@ Metrics will be available at `http://localhost:2112/metrics` - `--celestia-token`: Celestia authentication token (optional), sent as a `Bearer` token on all Celestia requests, including block timestamp lookups. Required when the celestia-node has auth enabled (the default). - `--duration`: Duration in seconds to stream (0 = infinite) - `--chain-id`: Chain identifier for metrics labels (default: "testnet") -- `--enable-metrics`: Enable Prometheus metrics HTTP server (default: false) - `--port`: HTTP server port for metrics (default: 2112) - `--jsonrpc-scrape-interval`: JSON-RPC health check scrape interval in seconds (default: 10) - `--reference-node`: Reference node RPC endpoint URL (sequencer) for drift monitoring @@ -76,8 +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) -- `--verbose`: Enable verbose logging (default: false) -- `--debug`: Enable debug logging for submission details (default: false, can also be set via `EVMETRICS_DEBUG=true` environment variable) +- `--verbose`: Enable verbose (debug level) logging (default: false) ### Example with Custom Endpoints @@ -100,7 +99,6 @@ Enable JSON-RPC request duration monitoring by providing the `--evm-rpc-url` fla --data-namespace collect_testnet_data \ --evm-ws-url "ws://localhost:8546" \ --evm-rpc-url "http://localhost:8545" \ - --enable-metrics \ --jsonrpc-scrape-interval 10 ``` @@ -116,15 +114,14 @@ Monitor native token balances for one or more Celestia addresses: --data-namespace collect_testnet_data \ --balance.addresses "celestia1abc...,celestia1def..." \ --balance.consensus-rpc-urls "https://rpc.celestia.org,https://rpc-mocha.pops.one" \ - --balance.scrape-interval 30 \ - --enable-metrics + --balance.scrape-interval 30 ``` This will query account balances every 30 seconds with automatic failover between RPC endpoints. ## Prometheus Metrics -When metrics are enabled, the following metrics are exposed: +The following metrics are exposed: ### `ev_metrics_unsubmitted_block_range_start` - **Type**: Gauge @@ -194,7 +191,7 @@ When metrics are enabled, the following metrics are exposed: ### `ev_metrics_block_time_slo_seconds` - **Type**: Gauge - **Labels**: `chain_id`, `quantile` -- **Description**: SLO thresholds for block time +- **Description**: SLO thresholds for block time (not yet emitted, see #44) - **Values**: - `0.5`: 2.0s - `0.9`: 3.0s @@ -204,13 +201,13 @@ When metrics are enabled, 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 +- **Description**: Delay between block creation and reception with histogram buckets (not yet emitted, see #44) - **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 +- **Description**: SLO thresholds for block receive delay (not yet emitted, see #44) - **Values**: - `0.5`: 1.0s - `0.9`: 3.0s @@ -277,11 +274,14 @@ When `--balance.addresses` and `--balance.consensus-rpc-urls` are provided: ## Debug Logging -Debug logging provides detailed visibility into DA submission process. Enable with `--debug` flag or `EVMETRICS_DEBUG=true` environment variable. +There are two independent debug logging mechanisms: + +- `--verbose` flag: sets the application log level to debug. +- `EVMETRICS_DEBUG=true` environment variable: logs DA submission details from the metrics package. -**Use cases**: Troubleshoot submission failures, verify submission flow, diagnose Celestia RPC issues. +**Use cases**: Troubleshoot submission failures, verify submission flow. -**Logs include**: Successful/failed submissions, DA height updates, submission timing. +**`EVMETRICS_DEBUG` logs include**: Successful/failed submission attempts and DA height updates. **Example**: ```bash diff --git a/cmd/monitor.go b/cmd/monitor.go index ab61793..f45835f 100644 --- a/cmd/monitor.go +++ b/cmd/monitor.go @@ -92,7 +92,7 @@ func NewMonitorCmd() *cobra.Command { cmd.Flags().StringVar(&flags.dataNS, flagDataNS, "", "Data namespace (my_app_data_namespace)") cmd.Flags().IntVar(&flags.duration, flagDuration, 0, "Duration in seconds to stream (0 = infinite)") cmd.Flags().BoolVar(&flags.verbose, flagVerbose, false, "Enable verbose logging") - cmd.Flags().IntVar(&flags.port, flagPort, 2112, "HTTP server port for metrics (only used if --enable-metrics is set)") + cmd.Flags().IntVar(&flags.port, flagPort, 2112, "HTTP server port for metrics") cmd.Flags().StringVar(&flags.chainID, flagChain, "testnet", "chainID identifier for metrics labels") cmd.Flags().StringVar(&flags.referenceNode, flagReferenceNode, "", "Reference node RPC endpoint URL (sequencer) for drift monitoring") cmd.Flags().StringVar(&flags.fullNodes, flagFullNodes, "", "Comma-separated list of full node RPC endpoint URLs for drift monitoring") @@ -248,8 +248,8 @@ func monitorAndExportMetrics(_ *cobra.Command, _ []string) error { err = metrics.StartServer(ctx, metricsPath, flags.port, logger, exporters...) - // context.Canceled is expected during graceful shutdown, not an error - if err != nil && !errors.Is(err, context.Canceled) { + // context.Canceled (shutdown) and context.DeadlineExceeded (--duration) are clean exits + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { return err } diff --git a/pkg/exporters/drift/drift.go b/pkg/exporters/drift/drift.go index f4c5ad0..02a595c 100644 --- a/pkg/exporters/drift/drift.go +++ b/pkg/exporters/drift/drift.go @@ -30,11 +30,12 @@ type exporter struct { fullNodes []string pollingInterval int logger zerolog.Logger + lastHeight uint64 } // ExportMetrics continuously checks block heights of a reference node and multiple full nodes, // recording the heights and their drift into the provided metrics instance at specified intervals. -func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error { +func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error { ticker := time.NewTicker(time.Duration(e.pollingInterval) * time.Second) defer ticker.Stop() @@ -63,7 +64,10 @@ func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error { m.RecordReferenceBlockHeight(e.chainID, e.referenceNode, 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 - m.UpdateLastBlockTime(e.chainID, time.Now()) + if refHeight > e.lastHeight { + m.UpdateLastBlockTime(e.chainID, time.Now()) + e.lastHeight = refHeight + } e.logger.Info().Uint64("height", refHeight).Str("endpoint", e.referenceNode).Msg("recorded reference node height") // get each full node height and calculate drift @@ -80,10 +84,6 @@ func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error { m.RecordCurrentBlockHeight(e.chainID, fullNode, currentHeight) m.RecordBlockHeightDrift(e.chainID, fullNode, refHeight, currentHeight) - // 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 - m.UpdateLastBlockTime(e.chainID, time.Now()) - drift := int64(refHeight) - int64(currentHeight) e.logger.Info(). Uint64("ref_height", refHeight). diff --git a/pkg/exporters/drift/drift_test.go b/pkg/exporters/drift/drift_test.go new file mode 100644 index 0000000..15bc635 --- /dev/null +++ b/pkg/exporters/drift/drift_test.go @@ -0,0 +1,82 @@ +package drift + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/evstack/ev-metrics/pkg/metrics" + "github.com/prometheus/client_golang/prometheus" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +func newBlockNumberServer(t *testing.T, height *atomic.Uint64) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + ID json.RawMessage `json:"id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":"0x%x"}`, req.ID, height.Load()) + })) + t.Cleanup(srv.Close) + return srv +} + +func TestExportMetrics_TimeSinceLastBlockOnlyResetsWhenHeightAdvances(t *testing.T) { + var height atomic.Uint64 + height.Store(10) + srv := newBlockNumberServer(t, &height) + + reg := prometheus.NewRegistry() + m := metrics.NewWithRegistry("test", reg) + e := NewMetricsExporter("chain", srv.URL, nil, 1, zerolog.Nop()) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = e.ExportMetrics(ctx, m) + }() + t.Cleanup(func() { + cancel() + <-done + }) + + timeSince := func() float64 { + m.UpdateTimeSinceLastBlock() + families, err := reg.Gather() + require.NoError(t, err) + for _, mf := range families { + if mf.GetName() == "test_time_since_last_block_seconds" { + return mf.GetMetric()[0].GetGauge().GetValue() + } + } + t.Fatal("time_since_last_block_seconds not found") + return 0 + } + + // polls at ~1s (height 10, first seen) and ~2s (height unchanged) + time.Sleep(2500 * time.Millisecond) + require.GreaterOrEqual(t, timeSince(), 1.2, "constant height must not reset the gauge") + + // poll at ~3s sees a lower height + height.Store(5) + time.Sleep(time.Second) + require.GreaterOrEqual(t, timeSince(), 2.2, "decreasing height must not reset the gauge") + + // poll at ~4s sees a new height + height.Store(11) + time.Sleep(time.Second) + require.Less(t, timeSince(), 1.0, "advancing height must reset the gauge") +} diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 537797e..2b65fed 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -86,7 +86,6 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error // It captures sub by reference so reassignment during reconnection is reflected. shutdown := func() { sub.Unsubscribe() - close(blockQueue) workerGroup.Wait() } @@ -108,7 +107,6 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error newSub := e.reconnectSubscription(ctx, headers) if newSub == nil { // context was cancelled during reconnection - close(blockQueue) workerGroup.Wait() return nil } @@ -173,26 +171,32 @@ func (e *exporter) processBlocks(ctx context.Context, m *metrics.Metrics, worker logger := e.logger.With().Int("worker_id", workerID).Logger() logger.Debug().Msg("worker started") - for header := range blockQueue { - reEnqueue := e.verifyBlock(ctx, m, header) - if reEnqueue { - // re-queue with delay in background - go func(h *types.Header) { - select { - case <-time.After(5 * time.Minute): + // blockQueue is never closed: delayed re-queue goroutines may still send to it, + // so workers exit on ctx cancellation instead. + for { + select { + case <-ctx.Done(): + logger.Debug().Msg("worker stopped") + return + case header := <-blockQueue: + reEnqueue := e.verifyBlock(ctx, m, header) + if reEnqueue { + // re-queue with delay in background + go func(h *types.Header) { select { - case blockQueue <- h: - logger.Debug().Uint64("block", h.Number.Uint64()).Msg("re-queued block after cooldown") + case <-time.After(5 * time.Minute): + select { + case blockQueue <- h: + logger.Debug().Uint64("block", h.Number.Uint64()).Msg("re-queued block after cooldown") + case <-ctx.Done(): + // graceful shutdown, don't send + } case <-ctx.Done(): - // graceful shutdown, don't send } - case <-ctx.Done(): - } - }(header) + }(header) + } } } - - logger.Debug().Msg("worker stopped") } func (e *exporter) onVerified(m *metrics.Metrics, namespace string, blockHeight, daHeight uint64, verified bool, submissionDuration time.Duration) { diff --git a/pkg/exporters/verifier/verifier_test.go b/pkg/exporters/verifier/verifier_test.go new file mode 100644 index 0000000..ec59afe --- /dev/null +++ b/pkg/exporters/verifier/verifier_test.go @@ -0,0 +1,30 @@ +package verifier + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/rs/zerolog" +) + +func TestProcessBlocksStopsOnContextCancel(t *testing.T) { + e := &exporter{logger: zerolog.Nop()} + queue := make(chan *types.Header, 1) + ctx, cancel := context.WithCancel(context.Background()) + + done := make(chan struct{}) + go func() { + e.processBlocks(ctx, nil, 0, queue) + close(done) + }() + + cancel() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("processBlocks did not return after context cancellation") + } +} diff --git a/pkg/metrics/server.go b/pkg/metrics/server.go index c5f141b..e023bf3 100644 --- a/pkg/metrics/server.go +++ b/pkg/metrics/server.go @@ -33,7 +33,8 @@ func StartServer(ctx context.Context, metricsPath string, port int, logger zerol // initialize metrics (thread-safe, shared across all exporters) m := New("ev_metrics") - var g errgroup.Group + // derived ctx is cancelled when any exporter or the server fails, stopping everything else + g, ctx := errgroup.WithContext(ctx) // start all metric exporters for _, exporter := range exporters { diff --git a/pkg/metrics/server_test.go b/pkg/metrics/server_test.go new file mode 100644 index 0000000..8413000 --- /dev/null +++ b/pkg/metrics/server_test.go @@ -0,0 +1,36 @@ +package metrics + +import ( + "context" + "errors" + "net" + "testing" + "time" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +type failingExporter struct{ err error } + +func (f failingExporter) ExportMetrics(context.Context, *Metrics) error { return f.err } + +func TestStartServer_ExporterErrorStopsServer(t *testing.T) { + l, err := net.Listen("tcp", ":0") + require.NoError(t, err) + port := l.Addr().(*net.TCPAddr).Port + require.NoError(t, l.Close()) + + wantErr := errors.New("subscribe failed") + done := make(chan error, 1) + go func() { + done <- StartServer(context.Background(), "/metrics", port, zerolog.Nop(), failingExporter{err: wantErr}) + }() + + select { + case err := <-done: + require.ErrorIs(t, err, wantErr) + case <-time.After(10 * time.Second): + t.Fatal("StartServer did not return after exporter failure") + } +}