Monitoring service api (sigp#2251)

## Issue Addressed

N/A

## Proposed Changes

Adds a client side api for collecting system and process metrics and pushing it to a monitoring service.

Author: pawanjay176

Cargo.lock

@@ -40,6 +40,7 @@ members = [
     "common/validator_dir",
     "common/warp_utils",
     "common/fallback",
+    "common/monitoring_api",
 
     "consensus/cached_tree_hash",
     "consensus/int_to_bytes",

Cargo.toml

@@ -44,4 +44,5 @@ hyper = "0.14.4"
 lighthouse_version = { path = "../common/lighthouse_version" }
 hex = "0.4.2"
 slasher = { path = "../slasher" }
+monitoring_api = { path = "../common/monitoring_api" }
 sensitive_url = { path = "../common/sensitive_url" }

beacon_node/Cargo.toml

@@ -44,3 +44,4 @@ http_api = { path = "../http_api" }
 http_metrics = { path = "../http_metrics" }
 slasher = { path = "../../slasher" }
 slasher_service = { path = "../../slasher/service" }
+monitoring_api = {path = "../../common/monitoring_api"}

beacon_node/client/Cargo.toml

@@ -14,6 +14,7 @@ use environment::RuntimeContext;
 use eth1::{Config as Eth1Config, Service as Eth1Service};
 use eth2_libp2p::NetworkGlobals;
 use genesis::{interop_genesis_state, Eth1GenesisService};
+use monitoring_api::{MonitoringHttpClient, ProcessType};
 use network::{NetworkConfig, NetworkMessage, NetworkService};
 use slasher::Slasher;
 use slasher_service::SlasherService;
@@ -374,6 +375,22 @@ where
         SlasherService::new(beacon_chain, network_send).run(&context.executor)
     }
 
+    /// Start the explorer client which periodically sends beacon
+    /// and system metrics to the configured endpoint.
+    pub fn monitoring_client(self, config: &monitoring_api::Config) -> Result<Self, String> {
+        let context = self
+            .runtime_context
+            .as_ref()
+            .ok_or("monitoring_client requires a runtime_context")?
+            .service_context("monitoring_client".into());
+        let monitoring_client = MonitoringHttpClient::new(config, context.log().clone())?;
+        monitoring_client.auto_update(
+            context.executor,
+            vec![ProcessType::BeaconNode, ProcessType::System],
+        );
+        Ok(self)
+    }
+
     /// Immediately starts the service that periodically logs information each slot.
     pub fn notifier(self) -> Result<Self, String> {
         let context = self

beacon_node/client/src/builder.rs

@@ -66,6 +66,7 @@ pub struct Config {
     pub eth1: eth1::Config,
     pub http_api: http_api::Config,
     pub http_metrics: http_metrics::Config,
+    pub monitoring_api: Option<monitoring_api::Config>,
     pub slasher: Option<slasher::Config>,
 }
 
@@ -87,6 +88,7 @@ impl Default for Config {
             graffiti: Graffiti::default(),
             http_api: <_>::default(),
             http_metrics: <_>::default(),
+            monitoring_api: None,
             slasher: None,
             validator_monitor_auto: false,
             validator_monitor_pubkeys: vec![],

beacon_node/client/src/config.rs

@@ -6,4 +6,14 @@ lazy_static! {
         "sync_slots_per_second",
         "The number of blocks being imported per second"
     );
+
+    pub static ref IS_SYNCED: Result<IntGauge> = try_create_int_gauge(
+        "sync_eth2_synced",
+        "Metric to check if the beacon chain is synced to head. 0 if not synced and non-zero if synced"
+    );
+
+    pub static ref NOTIFIER_HEAD_SLOT: Result<IntGauge> = try_create_int_gauge(
+        "notifier_head_slot",
+        "The head slot sourced from the beacon chain notifier"
+    );
 }

beacon_node/client/src/metrics.rs

@@ -77,6 +77,9 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
             };
 
             let head_slot = head_info.slot;
+
+            metrics::set_gauge(&metrics::NOTIFIER_HEAD_SLOT, head_slot.as_u64() as i64);
+
             let current_slot = match beacon_chain.slot() {
                 Ok(slot) => slot,
                 Err(e) => {
@@ -123,6 +126,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
 
             // Log if we are syncing
             if sync_state.is_syncing() {
+                metrics::set_gauge(&metrics::IS_SYNCED, 0);
                 let distance = format!(
                     "{} slots ({})",
                     head_distance.as_u64(),
@@ -151,6 +155,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
                     );
                 }
             } else if sync_state.is_synced() {
+                metrics::set_gauge(&metrics::IS_SYNCED, 1);
                 let block_info = if current_slot > head_slot {
                     "   …  empty".to_string()
                 } else {
@@ -167,6 +172,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
                     "slot" => current_slot,
                 );
             } else {
+                metrics::set_gauge(&metrics::IS_SYNCED, 0);
                 info!(
                     log,
                     "Searching for peers";

beacon_node/client/src/notifier.rs

@@ -26,4 +26,23 @@ lazy_static! {
     pub static ref ENDPOINT_REQUESTS: Result<IntCounterVec> = try_create_int_counter_vec(
         "eth1_endpoint_requests", "The number of eth1 requests for each endpoint", &["endpoint"]
     );
+
+    /*
+     * Eth1 rpc connection
+     */
+
+    pub static ref ETH1_CONNECTED: Result<IntGauge> = try_create_int_gauge(
+        "sync_eth1_connected", "Set to 1 if connected to an eth1 node, otherwise set to 0"
+    );
+
+    pub static ref ETH1_FALLBACK_CONFIGURED: Result<IntGauge> = try_create_int_gauge(
+        "sync_eth1_fallback_configured", "Number of configured eth1 fallbacks"
+    );
+
+    // Note: This metric only checks if an eth1 fallback is configured, not if it is connected and synced.
+    // Checking for liveness of the fallback would require moving away from lazy checking of fallbacks.
+    pub static ref ETH1_FALLBACK_CONNECTED: Result<IntGauge> = try_create_int_gauge(
+        "eth1_sync_fallback_connected", "Set to 1 if an eth1 fallback is connected, otherwise set to 0"
+    );
+
 }

beacon_node/eth1/src/metrics.rs

@@ -94,6 +94,9 @@ impl EndpointsCache {
                 &crate::metrics::ENDPOINT_ERRORS,
                 &[&endpoint.0.to_string()],
             );
+            crate::metrics::set_gauge(&metrics::ETH1_CONNECTED, 0);
+        } else {
+            crate::metrics::set_gauge(&metrics::ETH1_CONNECTED, 1);
         }
         state
     }
@@ -730,13 +733,23 @@ impl Service {
 
         let mut interval = interval_at(Instant::now(), update_interval);
 
+        let num_fallbacks = self.config().endpoints.len() - 1;
         let update_future = async move {
             loop {
                 interval.tick().await;
                 self.do_update(update_interval).await.ok();
             }
         };
 
+        // Set the number of configured eth1 servers
+        metrics::set_gauge(&metrics::ETH1_FALLBACK_CONFIGURED, num_fallbacks as i64);
+        // Since we lazily update eth1 fallbacks, it's not possible to know connection status of fallback.
+        // Hence, we set it to 1 if we have atleast one configured fallback.
+        if num_fallbacks > 0 {
+            metrics::set_gauge(&metrics::ETH1_FALLBACK_CONNECTED, 1);
+        } else {
+            metrics::set_gauge(&metrics::ETH1_FALLBACK_CONNECTED, 0);
+        }
         handle.spawn(update_future, "eth1");
     }