Move notifier and latency service to validator_services

validator_client/validator_services/src/latency_service.rs

@@ -0,0 +1,61 @@
+use beacon_node_fallback::BeaconNodeFallback;
+use slot_clock::SlotClock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
+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>(
+    executor: TaskExecutor,
+    slot_clock: T,
+    beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
+) {
+    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,
+                        );
+                    }
+                }
+            }
+        }
+    };
+
+    executor.spawn(future, "latency");
+}

validator_client/validator_services/src/lib.rs

@@ -1,6 +1,8 @@
 pub mod attestation_service;
 pub mod block_service;
 pub mod duties_service;
+pub mod latency_service;
+pub mod notifier_service;
 pub mod preparation_service;
 pub mod sync;
 pub mod sync_committee_service;

validator_client/validator_services/src/notifier_service.rs

@@ -0,0 +1,152 @@
+use crate::duties_service::DutiesService;
+use slot_clock::SlotClock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
+use tokio::time::{sleep, Duration};
+use tracing::{debug, error, info};
+use types::{ChainSpec, EthSpec};
+use validator_metrics::set_gauge;
+
+/// Spawns a notifier service which periodically logs information about the node.
+pub fn spawn_notifier<T: SlotClock + 'static, E: EthSpec>(
+    duties_service: Arc<DutiesService<T, E>>,
+    executor: TaskExecutor,
+    spec: &ChainSpec,
+) -> Result<(), String> {
+    let slot_duration = Duration::from_secs(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<T, E>) {
+    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");
+    }
+}