gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-19 21:04:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'separate-vc-metrics' into into-anchor

Browse Source
This commit is contained in:
Daniel Knopik
2025-05-13 11:05:14 +02:00
parent 99057a52bb ed8b939ac2
commit 0bc38e414e
5 changed files with 35 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
validator_client/src/latency.rs
View File

@@ -1,61 +0,0 @@
use beacon_node_fallback::BeaconNodeFallback;
use environment::RuntimeContext;
use slot_clock::SlotClock;
use std::sync::Arc;
use tokio::time::sleep;
use tracing::debug;
use types::EthSpec;
/// The latency service will run 11/12ths of the way through the slot.
pub const SLOT_DELAY_MULTIPLIER: u32 = 11;
pub const SLOT_DELAY_DENOMINATOR: u32 = 12;
/// Starts a service that periodically checks the latency between the VC and the
/// candidate BNs.
pub fn start_latency_service<T: SlotClock + 'static, E: EthSpec>(
context: RuntimeContext<E>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T>>,
) {
let future = async move {
loop {
let sleep_time = slot_clock
.duration_to_next_slot()
.map(|next_slot| {
// This is 11/12ths through the next slot. On mainnet this
// will happen in the 11th second of each slot, one second
// before the next slot.
next_slot + (next_slot / SLOT_DELAY_DENOMINATOR) * SLOT_DELAY_MULTIPLIER
})
// If we can't read the slot clock, just wait one slot. Running
// the measurement at a non-exact time is not a big issue.
.unwrap_or_else(|| slot_clock.slot_duration());
// Sleep until it's time to perform the measurement.
sleep(sleep_time).await;
for (i, measurement) in beacon_nodes.measure_latency().await.iter().enumerate() {
if let Some(latency) = measurement.latency {
debug!(
node = &measurement.beacon_node_id,
latency = latency.as_millis(),
"Measured BN latency"
);
validator_metrics::observe_timer_vec(
&validator_metrics::VC_BEACON_NODE_LATENCY,
&[&measurement.beacon_node_id],
latency,
);
if i == 0 {
validator_metrics::observe_duration(
&validator_metrics::VC_BEACON_NODE_LATENCY_PRIMARY_ENDPOINT,
latency,
);
}
}
}
}
};
context.executor.spawn(future, "latency");
}

17
validator_client/src/lib.rs
View File

@@ -1,7 +1,5 @@
pub mod cli;
pub mod config;
mod latency;
mod notifier;
use crate::cli::ValidatorClient;
use crate::duties_service::SelectionProofConfig;
@@ -22,7 +20,6 @@ use environment::RuntimeContext;
use eth2::{reqwest::ClientBuilder, BeaconNodeHttpClient, StatusCode, Timeouts};
use initialized_validators::Error::UnableToOpenVotingKeystore;
use lighthouse_validator_store::LighthouseValidatorStore;
use notifier::spawn_notifier;
use parking_lot::RwLock;
use reqwest::Certificate;
use slot_clock::SlotClock;
@@ -40,10 +37,12 @@ use tokio::{
use tracing::{debug, error, info, warn};
use types::{EthSpec, Hash256};
use validator_http_api::ApiSecret;
use validator_services::notifier_service::spawn_notifier;
use validator_services::{
attestation_service::{AttestationService, AttestationServiceBuilder},
block_service::{BlockService, BlockServiceBuilder},
duties_service::{self, DutiesService, DutiesServiceBuilder},
latency_service,
preparation_service::{PreparationService, PreparationServiceBuilder},
sync_committee_service::SyncCommitteeService,
};
@@ -659,11 +658,17 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
info!("Doppelganger protection disabled.")
}
spawn_notifier(self).map_err(|e| format!("Failed to start notifier: {}", e))?;
let context = self.context.service_context("notifier".into());
spawn_notifier(
self.duties_service.clone(),
context.executor,
&self.context.eth2_config.spec,
)
.map_err(|e| format!("Failed to start notifier: {}", e))?;
if self.config.enable_latency_measurement_service {
latency::start_latency_service(
self.context.clone(),
latency_service::start_latency_service(
self.context.executor.clone(),
self.duties_service.slot_clock.clone(),
self.duties_service.beacon_nodes.clone(),
);

153
validator_client/src/notifier.rs
View File

@@ -1,153 +0,0 @@
use crate::{DutiesService, ProductionValidatorClient};
use lighthouse_validator_store::LighthouseValidatorStore;
use metrics::set_gauge;
use slot_clock::SlotClock;
use tokio::time::{sleep, Duration};
use tracing::{debug, error, info};
use types::EthSpec;
/// Spawns a notifier service which periodically logs information about the node.
pub fn spawn_notifier<E: EthSpec>(client: &ProductionValidatorClient<E>) -> Result<(), String> {
let context = client.context.service_context("notifier".into());
let executor = context.executor.clone();
let duties_service = client.duties_service.clone();
let slot_duration = Duration::from_secs(context.eth2_config.spec.seconds_per_slot);
let interval_fut = async move {
loop {
if let Some(duration_to_next_slot) = duties_service.slot_clock.duration_to_next_slot() {
sleep(duration_to_next_slot + slot_duration / 2).await;
notify(&duties_service).await;
} else {
error!("Failed to read slot clock");
// If we can't read the slot clock, just wait another slot.
sleep(slot_duration).await;
continue;
}
}
};
executor.spawn(interval_fut, "validator_notifier");
Ok(())
}
/// Performs a single notification routine.
async fn notify<T: SlotClock + 'static, E: EthSpec>(
duties_service: &DutiesService<LighthouseValidatorStore<T, E>, T>,
) {
let (candidate_info, num_available, num_synced) =
duties_service.beacon_nodes.get_notifier_info().await;
let num_total = candidate_info.len();
let num_synced_fallback = num_synced.saturating_sub(1);
set_gauge(
&validator_metrics::AVAILABLE_BEACON_NODES_COUNT,
num_available as i64,
);
set_gauge(
&validator_metrics::SYNCED_BEACON_NODES_COUNT,
num_synced as i64,
);
set_gauge(
&validator_metrics::TOTAL_BEACON_NODES_COUNT,
num_total as i64,
);
if num_synced > 0 {
let primary = candidate_info
.first()
.map(|candidate| candidate.endpoint.as_str())
.unwrap_or("None");
info!(
primary,
total = num_total,
available = num_available,
synced = num_synced,
"Connected to beacon node(s)"
)
} else {
error!(
total = num_total,
available = num_available,
synced = num_synced,
"No synced beacon nodes"
)
}
if num_synced_fallback > 0 {
set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 1);
} else {
set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 0);
}
for info in candidate_info {
if let Ok(health) = info.health {
debug!(
status = "Connected",
index = info.index,
endpoint = info.endpoint,
head_slot = %health.head,
is_optimistic = ?health.optimistic_status,
execution_engine_status = ?health.execution_status,
health_tier = %health.health_tier,
"Beacon node info"
);
} else {
debug!(
status = "Disconnected",
index = info.index,
endpoint = info.endpoint,
"Beacon node info"
);
}
}
if let Some(slot) = duties_service.slot_clock.now() {
let epoch = slot.epoch(E::slots_per_epoch());
let total_validators = duties_service.total_validator_count();
let proposing_validators = duties_service.proposer_count(epoch);
let attesting_validators = duties_service.attester_count(epoch);
let doppelganger_detecting_validators = duties_service.doppelganger_detecting_count();
if doppelganger_detecting_validators > 0 {
info!(
doppelganger_detecting_validators,
"Listening for doppelgangers"
)
}
if total_validators == 0 {
info!(
msg = "see `lighthouse vm create --help` or the HTTP API documentation",
"No validators present"
)
} else if total_validators == attesting_validators {
info!(
current_epoch_proposers = proposing_validators,
active_validators = attesting_validators,
total_validators = total_validators,
%epoch,
%slot,
"All validators active"
);
} else if attesting_validators > 0 {
info!(
current_epoch_proposers = proposing_validators,
active_validators = attesting_validators,
total_validators = total_validators,
%epoch,
%slot,
"Some validators active"
);
} else {
info!(
validators = total_validators,
%epoch,
%slot,
"Awaiting activation"
);
}
} else {
error!("Unable to read slot clock");
}
}