Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion bin/debug-trace-server/src/data_provider.rs
Original file line number Diff line number Diff line change
Expand Up @@ -767,7 +767,7 @@ async fn resolve_contracts_inner(
"Cache miss — fetching contracts from RPC"
);

// Per-attempt `eth_getCodeByHash` metrics land on `UpstreamMetrics` via the
// Per-attempt `eth_getCodeByHash` metrics land on the upstream attempt metrics via the
// `TraceRpcMetrics` adapter inside `round_robin_with_backoff`.
let fetched = rpc_client.get_codes_with_deadline(&missing, true, Some(deadline)).await?;

Expand Down
107 changes: 58 additions & 49 deletions bin/debug-trace-server/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
use std::net::SocketAddr;

use eyre::Result;
use metrics::{Counter, Gauge, Histogram};
use metrics::{Counter, Gauge, Histogram, counter, histogram};
use metrics_derive::Metrics;
use metrics_exporter_prometheus::{Matcher, PrometheusBuilder};
pub use stateless_common::{
Expand Down Expand Up @@ -246,32 +246,33 @@ impl SingleFlightMetrics {
}
}

/// Upstream RPC metrics with method label.
#[derive(Clone, Metrics)]
#[metrics(scope = "debug_trace")]
pub struct UpstreamMetrics {
/// Total upstream RPC requests
upstream_requests_total: Counter,
/// Total upstream RPC errors
upstream_errors_total: Counter,
/// Duration of upstream RPC requests in seconds
upstream_duration_seconds: Histogram,
}
/// Total upstream RPC attempts, labeled `(method, provider, outcome)`.
const UPSTREAM_REQUESTS_TOTAL: &str = "debug_trace_upstream_requests_total";
/// Per-attempt upstream RPC duration in seconds, labeled `(method, provider, outcome)`.
const UPSTREAM_DURATION_SECONDS: &str = "debug_trace_upstream_duration_seconds";
/// Logical-call deadline-exceeded ("request timed out") count, labeled `(method)`.
const UPSTREAM_DEADLINE_EXCEEDED_TOTAL: &str = "debug_trace_upstream_deadline_exceeded_total";

impl UpstreamMetrics {
/// Creates metrics for a specific upstream RPC method.
pub fn new_for_method(method: &'static str) -> Self {
Self::new_with_labels(&[("method", method)])
}
/// Records one upstream RPC attempt against `provider` with its `outcome` and duration.
///
/// The `(method, provider, outcome)` labeling is what lets operators attribute latency and
/// separate a returned error from a stall-timeout, per endpoint — the collapsed method-only
/// `success` boolean could do neither. `provider` is a bounded, credential-free endpoint host.
fn record_upstream_attempt(
method: &'static str,
provider: &str,
outcome: &'static str,
duration_secs: f64,
) {
counter!(UPSTREAM_REQUESTS_TOTAL, "method" => method, "provider" => provider.to_owned(), "outcome" => outcome)
.increment(1);
histogram!(UPSTREAM_DURATION_SECONDS, "method" => method, "provider" => provider.to_owned(), "outcome" => outcome)
.record(duration_secs);
}

/// Records an upstream RPC request.
pub fn record_request(&self, success: bool, duration_secs: f64) {
self.upstream_requests_total.increment(1);
if !success {
self.upstream_errors_total.increment(1);
}
self.upstream_duration_seconds.record(duration_secs);
}
/// Records one logical upstream call giving up because its overall deadline elapsed.
fn record_upstream_deadline_exceeded(method: &'static str) {
counter!(UPSTREAM_DEADLINE_EXCEEDED_TOTAL, "method" => method).increment(1);
}

/// Witness fetch metrics by source.
Expand Down Expand Up @@ -426,11 +427,15 @@ fn pre_register_all_metrics() {
let _ = SingleFlightMetrics::new_for_type("coalesced");
let _ = SingleFlightMetrics::new_for_type("bypassed");

// Data Fetch Layer: upstream RPC
let _ = UpstreamMetrics::new_for_method("eth_getHeaderByHash");
let _ = UpstreamMetrics::new_for_method("eth_getBlockByHash");
let _ = UpstreamMetrics::new_for_method("mega_getWitness");
let _ = UpstreamMetrics::new_for_method("eth_getCodeByHash");
// Data Fetch Layer: upstream RPC. The attempt series (`upstream_requests_total` /
// `upstream_duration_seconds`) carry a dynamic per-endpoint `provider` label, so they first
// appear on the initial attempt; only the method-keyed deadline-exceeded counter is
// pre-registrable here.
for method in
["eth_getHeaderByHash", "eth_getBlockByHash", "mega_getWitness", "eth_getCodeByHash"]
{
counter!(UPSTREAM_DEADLINE_EXCEEDED_TOTAL, "method" => method).increment(0);
}

// Witness Layer
let _ = WitnessSourceMetrics::new_for_source("witness_generator");
Expand Down Expand Up @@ -495,7 +500,7 @@ pub fn record_rpc_error(method: &str) {
RpcMethodMetrics::new_for_method(method).record_error();
}

/// Maps an [`RpcMethod`] to the `method` label used by [`UpstreamMetrics`].
/// Maps an [`RpcMethod`] to the `method` label used by the upstream attempt metrics.
///
/// The existing dashboard labels (`eth_getHeaderByHash`, `eth_getBlockByHash`, etc.)
/// encode the trace-server-specific call flavor. Since [`RpcMethod`] is coarser
Expand All @@ -513,38 +518,42 @@ fn upstream_label_for(method: stateless_common::metrics::RpcMethod) -> &'static
}

/// [`stateless_common::RpcMetrics`] adapter that forwards every per-attempt RPC
/// event to [`UpstreamMetrics`], keyed by method label.
/// event to the upstream metrics, keyed by `(method, provider, outcome)`.
///
/// Wired via [`stateless_common::RpcClientConfig::with_metrics`] so that the
/// per-attempt duration and success/failure counters recorded inside
/// `round_robin_with_backoff` land on the same dashboards the trace server has
/// always exposed — even though `get_block` / `get_header` / `get_witness` no
/// longer surface their per-call success status at the caller level.
/// per-attempt duration and per-endpoint outcome counters recorded inside
/// `round_robin_with_backoff` reach Prometheus — including the primary-vs-failover
/// split for `mega_getWitness` and the reason (error vs timeout) for each failure.
#[derive(Default)]
pub struct TraceRpcMetrics;

impl stateless_common::RpcMetrics for TraceRpcMetrics {
fn on_rpc_complete(
fn on_rpc_attempt(
&self,
method: stateless_common::metrics::RpcMethod,
success: bool,
duration_secs: Option<f64>,
provider: &str,
outcome: stateless_common::metrics::RpcAttemptOutcome,
duration_secs: f64,
) {
let label = upstream_label_for(method);
// `new_for_method` is just a label binding — no allocation beyond what
// the metrics crate deduplicates internally.
UpstreamMetrics::new_for_method(label)
.record_request(success, duration_secs.unwrap_or(0.0));
record_upstream_attempt(
upstream_label_for(method),
provider,
outcome.as_str(),
duration_secs,
);
}

fn on_rpc_retry(&self, _method: stateless_common::metrics::RpcMethod) {
// `on_rpc_complete(_, false, _)` fires alongside `on_rpc_retry` in the
// retry loop, so the error/request counters already capture retries.
fn on_rpc_deadline_exceeded(
&self,
method: stateless_common::metrics::RpcMethod,
_elapsed_secs: f64,
) {
record_upstream_deadline_exceeded(upstream_label_for(method));
}

fn on_witness_fetch(&self, _breakdown: stateless_common::witness_size::WitnessSizeBreakdown) {
// Witness size/source metrics are recorded by `fetch_witness_with_timeout`
// at the outer timeout boundary (distinct semantics from per-attempt).
// Witness size/source metrics are recorded by `fetch_witness` at the outer
// boundary (distinct semantics from per-attempt).
}
}

Expand Down
53 changes: 36 additions & 17 deletions bin/stateless-validator/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ use metrics::{counter, describe_counter, describe_gauge, describe_histogram, gau
use metrics_exporter_prometheus::{Matcher, PrometheusBuilder};
pub use stateless_common::{
DEFAULT_METRICS_PORT, WitnessSizeBreakdown,
metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcMethod, RpcMetrics},
metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcAttemptOutcome, RpcMethod, RpcMetrics},
};
use tracing::info;

Expand All @@ -27,14 +27,24 @@ use crate::r2_witness::R2WitnessError;
pub struct ValidatorMetrics;

impl RpcMetrics for ValidatorMetrics {
fn on_rpc_complete(&self, method: RpcMethod, success: bool, duration_secs: Option<f64>) {
on_rpc_complete(method, success, duration_secs);
fn on_rpc_attempt(
&self,
method: RpcMethod,
_provider: &str,
outcome: RpcAttemptOutcome,
duration_secs: f64,
) {
on_rpc_attempt(method, outcome, duration_secs);
}

fn on_rpc_retry(&self, method: RpcMethod) {
on_rpc_retry(method);
}

fn on_rpc_deadline_exceeded(&self, method: RpcMethod, _elapsed_secs: f64) {
on_rpc_deadline_exceeded(method);
}

fn on_witness_fetch(&self, breakdown: WitnessSizeBreakdown) {
on_witness_fetch(breakdown);
}
Expand Down Expand Up @@ -73,6 +83,7 @@ pub mod names {
metric!(RPC_REQUESTS_TOTAL, "rpc_requests_total");
metric!(RPC_ERRORS_TOTAL, "rpc_errors_total");
metric!(RPC_RETRY_ATTEMPTS_TOTAL, "rpc_retry_attempts_total");
metric!(RPC_DEADLINE_EXCEEDED_TOTAL, "rpc_deadline_exceeded_total");
metric!(BLOCK_FETCH_TIME, "block_fetch_time_seconds");
metric!(CODE_FETCH_TIME, "code_fetch_time_seconds");
metric!(WITNESS_FETCH_RPC_TIME, "witness_fetch_rpc_time_seconds");
Expand Down Expand Up @@ -163,6 +174,10 @@ fn register_metric_descriptions() {
names::RPC_RETRY_ATTEMPTS_TOTAL,
"RPC transient retry attempts (before final outcome)"
);
describe_counter!(
names::RPC_DEADLINE_EXCEEDED_TOTAL,
"Logical RPC calls that gave up because their overall deadline elapsed"
);
describe_histogram!(names::BLOCK_FETCH_TIME, "Block fetch time (s)");
describe_histogram!(names::CODE_FETCH_TIME, "Code fetch time (s)");
describe_histogram!(names::WITNESS_FETCH_RPC_TIME, "Witness RPC fetch time (s)");
Expand Down Expand Up @@ -209,6 +224,7 @@ fn init_rpc_method_counters() {
counter!(names::RPC_REQUESTS_TOTAL, "method" => method_str).increment(0);
counter!(names::RPC_ERRORS_TOTAL, "method" => method_str).increment(0);
counter!(names::RPC_RETRY_ATTEMPTS_TOTAL, "method" => method_str).increment(0);
counter!(names::RPC_DEADLINE_EXCEEDED_TOTAL, "method" => method_str).increment(0);
}
}

Expand Down Expand Up @@ -286,29 +302,32 @@ pub fn on_rpc_retry(method: RpcMethod) {
counter!(names::RPC_RETRY_ATTEMPTS_TOTAL, "method" => method.as_str()).increment(1);
}

pub fn on_rpc_complete(method: RpcMethod, success: bool, duration_secs: Option<f64>) {
pub fn on_rpc_attempt(method: RpcMethod, outcome: RpcAttemptOutcome, duration_secs: f64) {
let method_str = method.as_str();
counter!(names::RPC_REQUESTS_TOTAL, "method" => method_str).increment(1);
if !success {
if !outcome.is_success() {
counter!(names::RPC_ERRORS_TOTAL, "method" => method_str).increment(1);
}

if let Some(duration) = duration_secs {
match method {
RpcMethod::EthGetCodeByHash => {
histogram!(names::CODE_FETCH_TIME).record(duration);
}
RpcMethod::EthGetBlock => {
histogram!(names::BLOCK_FETCH_TIME).record(duration);
}
RpcMethod::MegaGetBlockWitness => {
histogram!(names::WITNESS_FETCH_RPC_TIME).record(duration);
}
_ => {}
match method {
RpcMethod::EthGetCodeByHash => {
histogram!(names::CODE_FETCH_TIME).record(duration_secs);
}
RpcMethod::EthGetBlock => {
histogram!(names::BLOCK_FETCH_TIME).record(duration_secs);
}
RpcMethod::MegaGetBlockWitness => {
histogram!(names::WITNESS_FETCH_RPC_TIME).record(duration_secs);
}
_ => {}
}
}

/// Record a logical RPC call giving up because its overall deadline elapsed.
pub fn on_rpc_deadline_exceeded(method: RpcMethod) {
counter!(names::RPC_DEADLINE_EXCEEDED_TOTAL, "method" => method.as_str()).increment(1);
}

pub fn on_contract_cache_read(hits: u64, misses: u64) {
if hits > 0 {
counter!(names::CONTRACT_CACHE_HITS).increment(hits);
Expand Down
72 changes: 69 additions & 3 deletions crates/stateless-common/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,18 +50,74 @@ impl RpcMethod {
}
}

/// Outcome of a single provider attempt inside the retry loop.
///
/// The retry loop reports one of these per provider round trip, so metrics can
/// attribute latency and failure *reason* per endpoint instead of collapsing
/// every non-success into one opaque error count. `Error` and `Timeout` are
/// both retriable failures; the split lets operators tell a provider that
/// answered with an error from one that stalled and had to be timed out.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RpcAttemptOutcome {
/// The provider returned a usable response.
Success,
/// The provider returned an error, or its response failed to decode.
Error,
/// The attempt hit the per-attempt timeout: the provider accepted the
/// request but did not answer within the budget (a stall).
Timeout,
}

impl RpcAttemptOutcome {
/// Stable metric-label string for this outcome.
pub fn as_str(&self) -> &'static str {
match self {
RpcAttemptOutcome::Success => "success",
RpcAttemptOutcome::Error => "error",
RpcAttemptOutcome::Timeout => "timeout",
}
}

/// Whether this attempt succeeded.
pub fn is_success(&self) -> bool {
matches!(self, RpcAttemptOutcome::Success)
}
}

/// Trait for RPC metrics callbacks.
///
/// Implement this trait to receive metrics events from the RPC client.
/// Implement this trait to receive metrics events from the RPC client. All
/// callbacks fire from inside the retry loop, so `on_rpc_attempt` is called
/// once per provider round trip (not once per logical call).
pub trait RpcMetrics: Send + Sync {
/// Called when an RPC request completes (final outcome — success or permanent failure).
fn on_rpc_complete(&self, method: RpcMethod, success: bool, duration_secs: Option<f64>);
/// Called once per provider attempt, tagged with the endpoint `provider`
/// label and the attempt [`RpcAttemptOutcome`].
///
/// Every round trip the retry loop makes fires exactly one of these, so
/// both per-endpoint latency (via `duration_secs`) and per-reason failure
/// counts (via `outcome`) are derivable. `provider` is a bounded,
/// credential-free endpoint label (see the RPC client's `endpoint_label`).
fn on_rpc_attempt(
&self,
method: RpcMethod,
provider: &str,
outcome: RpcAttemptOutcome,
duration_secs: f64,
);

/// Called on each transient failure that will be retried (not on the final outcome).
///
/// Default: no-op. Implement to track retry volume separately from logical errors.
fn on_rpc_retry(&self, _method: RpcMethod) {}

/// Called when a logical call gives up because its overall deadline elapsed.
///
/// Distinct from a per-attempt [`RpcAttemptOutcome::Timeout`]: this counts
/// the *logical call* exhausting its whole budget (e.g. a witness fetch's
/// 3s deadline), which is the operator-facing "request timed out" signal.
/// Fires at most once per logical call. Default: no-op.
fn on_rpc_deadline_exceeded(&self, _method: RpcMethod, _elapsed_secs: f64) {}

/// Called when witness data is successfully fetched.
fn on_witness_fetch(&self, breakdown: WitnessSizeBreakdown);
}
Expand All @@ -78,4 +134,14 @@ mod tests {
assert_eq!(RpcMethod::MegaGetBlockWitness.as_str(), "mega_getBlockWitness");
assert_eq!(RpcMethod::MegaSetValidatedBlocks.as_str(), "mega_setValidatedBlocks");
}

#[test]
fn test_rpc_attempt_outcome() {
assert_eq!(RpcAttemptOutcome::Success.as_str(), "success");
assert_eq!(RpcAttemptOutcome::Error.as_str(), "error");
assert_eq!(RpcAttemptOutcome::Timeout.as_str(), "timeout");
assert!(RpcAttemptOutcome::Success.is_success());
assert!(!RpcAttemptOutcome::Error.is_success());
assert!(!RpcAttemptOutcome::Timeout.is_success());
}
}
Loading
Loading