Skip to content

Commit ed2fea0

Browse files
feat(metrics): implement lean_aggregator_skipped_total from leanMetrics PR #36 (#430)
## 🗒️ Description / Motivation Implements the new cross-client counter `lean_aggregator_skipped_total{reason=...}` proposed in [leanMetrics PR #36](leanEthereum/leanMetrics#36), so operators can attribute missed aggregations directly instead of deriving them from coverage gauges or logs. ## What Changed | reason | When it fires in ethlambda | |---|---| | `not_aggregator` | Every interval-2 tick where the aggregator flag is off — separates "no duty" from genuine misses | | `other` | Aggregation jobs the worker never reached because the 750 ms session deadline (or actor shutdown) cancelled it, incremented by the number of dropped jobs | | `not_synced`, `missing_state`, `spawn_failed` | Never fire — ethlambda has no sync gate on aggregation, no per-target pre-state resolution, and `spawn_blocking` cannot fail to start. Seeded at zero so fleet dashboards see the full label set | - `crates/blockchain/src/metrics.rs`: `IntCounterVec` registration, reason-list const documenting the never-firing labels, two `inc_*` helpers, init seeding - `crates/blockchain/src/lib.rs`: count the skip at the interval-2 tick when not an aggregator - `crates/blockchain/src/aggregation.rs`: track attempted vs. total jobs in `run_aggregation_worker` and count the dropped remainder on cancellation ## ✅ Verification Checklist - [x] Ran `make fmt` — clean - [x] Ran `make lint` (clippy with `-D warnings`) — clean - [x] `cargo test -p ethlambda-blockchain --lib` — 29 passed Co-authored-by: Tomás Grüner <47506558+MegaRedHand@users.noreply.github.com>
1 parent 7ca0b5e commit ed2fea0

3 files changed

Lines changed: 71 additions & 3 deletions

File tree

crates/blockchain/src/aggregation.rs

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,11 +403,14 @@ pub(crate) fn run_aggregation_worker(
403403
let mut groups_aggregated = 0usize;
404404
let mut total_raw_sigs = 0usize;
405405
let mut total_children = 0usize;
406+
let jobs_total = snapshot.jobs.len();
407+
let mut jobs_attempted = 0usize;
406408

407409
for job in snapshot.jobs {
408410
if cancel.is_cancelled() {
409411
break;
410412
}
413+
jobs_attempted += 1;
411414

412415
let slot = job.slot;
413416
let raw_sigs = job.raw_ids.len();
@@ -450,6 +453,13 @@ pub(crate) fn run_aggregation_worker(
450453
}
451454
}
452455

456+
// Jobs the loop never reached (deadline cancellation or actor gone) are
457+
// skipped aggregation submissions per leanMetrics.
458+
let jobs_dropped = jobs_total - jobs_attempted;
459+
if jobs_dropped > 0 {
460+
metrics::inc_aggregator_skipped_other(jobs_dropped as u64);
461+
}
462+
453463
let _ = actor.send(AggregationDone {
454464
session_id,
455465
groups_considered,

crates/blockchain/src/lib.rs

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -254,9 +254,16 @@ impl BlockChainServer {
254254
proposer_validator_id.is_some(),
255255
);
256256

257-
if interval == 2 && is_aggregator {
258-
coverage::emit_agg_start_new_coverage(&self.store, self.attestation_committee_count);
259-
self.start_aggregation_session(slot, ctx).await;
257+
if interval == 2 {
258+
if is_aggregator {
259+
coverage::emit_agg_start_new_coverage(
260+
&self.store,
261+
self.attestation_committee_count,
262+
);
263+
self.start_aggregation_session(slot, ctx).await;
264+
} else {
265+
metrics::inc_aggregator_skipped_not_aggregator();
266+
}
260267
}
261268

262269
// Now build and publish the block (after attestations have been accepted)

crates/blockchain/src/metrics.rs

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -513,6 +513,33 @@ static LEAN_NODE_SYNC_STATUS: std::sync::LazyLock<IntGaugeVec> = std::sync::Lazy
513513
register_int_gauge_vec!("lean_node_sync_status", "Node sync status", &["status"]).unwrap()
514514
});
515515

516+
// --- Aggregator Skips ---
517+
518+
/// Cross-client label set for `lean_aggregator_skipped_total` (leanMetrics).
519+
///
520+
/// `not_synced`, `missing_state` and `spawn_failed` never fire in ethlambda
521+
/// today: aggregation is not gated on sync status, needs no per-target
522+
/// pre-state resolution, and the `spawn_blocking` worker cannot fail to
523+
/// start. They are seeded at zero so fleet-wide dashboards see the full
524+
/// series.
525+
const AGGREGATOR_SKIP_REASONS: &[&str] = &[
526+
"not_aggregator",
527+
"not_synced",
528+
"missing_state",
529+
"spawn_failed",
530+
"other",
531+
];
532+
533+
static LEAN_AGGREGATOR_SKIPPED_TOTAL: std::sync::LazyLock<IntCounterVec> =
534+
std::sync::LazyLock::new(|| {
535+
register_int_counter_vec!(
536+
"lean_aggregator_skipped_total",
537+
"Aggregation submissions skipped, by reason",
538+
&["reason"]
539+
)
540+
.unwrap()
541+
});
542+
516543
// --- Initialization ---
517544

518545
/// Register all metrics with the Prometheus registry so they appear in `/metrics` from startup.
@@ -588,6 +615,13 @@ pub fn init() {
588615
std::sync::LazyLock::force(&LEAN_BLOCK_PROPOSAL_AGGREGATES_SELECTED);
589616
// Sync status
590617
std::sync::LazyLock::force(&LEAN_NODE_SYNC_STATUS);
618+
// Aggregator skip counter: instantiate every cross-client reason so the
619+
// full series is visible from startup, including reasons ethlambda
620+
// never fires.
621+
std::sync::LazyLock::force(&LEAN_AGGREGATOR_SKIPPED_TOTAL);
622+
for &reason in AGGREGATOR_SKIP_REASONS {
623+
LEAN_AGGREGATOR_SKIPPED_TOTAL.with_label_values(&[reason]);
624+
}
591625
}
592626

593627
// --- Public API ---
@@ -725,6 +759,23 @@ pub fn observe_committee_signatures_aggregation(elapsed: std::time::Duration) {
725759
LEAN_COMMITTEE_SIGNATURES_AGGREGATION_TIME_SECONDS.observe(elapsed.as_secs_f64());
726760
}
727761

762+
/// One aggregation cycle (interval-2 tick) skipped because this node has no
763+
/// aggregation duty. Bookkeeping label that lets dashboards separate "no
764+
/// duty" from genuine misses.
765+
pub fn inc_aggregator_skipped_not_aggregator() {
766+
LEAN_AGGREGATOR_SKIPPED_TOTAL
767+
.with_label_values(&["not_aggregator"])
768+
.inc();
769+
}
770+
771+
/// Aggregation jobs dropped without being attempted, e.g. because the
772+
/// session deadline cancelled the worker before it reached them.
773+
pub fn inc_aggregator_skipped_other(count: u64) {
774+
LEAN_AGGREGATOR_SKIPPED_TOTAL
775+
.with_label_values(&["other"])
776+
.inc_by(count);
777+
}
778+
728779
/// Update a table byte size gauge.
729780
pub fn update_table_bytes(table_name: &str, bytes: u64) {
730781
LEAN_TABLE_BYTES

0 commit comments

Comments
 (0)