[Merged by Bors] - Monitoring service api

Cargo.toml

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

beacon_node/Cargo.toml

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

beacon_node/client/Cargo.toml

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

beacon_node/client/src/builder.rs

@@ -12,6 +12,7 @@ use beacon_chain::{
 use environment::RuntimeContext;
 use eth1::{Config as Eth1Config, Service as Eth1Service};
 use eth2_libp2p::NetworkGlobals;
+use explorer_api::{ExplorerHttpClient, ProcessType};
 use genesis::{interop_genesis_state, Eth1GenesisService};
 use network::{NetworkConfig, NetworkMessage, NetworkService};
 use slasher::Slasher;
@@ -375,6 +376,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 explorer_client(self, config: &explorer_api::Config) -> Result<Self, String> {
+        let context = self
+            .runtime_context
+            .as_ref()
+            .ok_or("explorer_client requires a runtime_context")?
+            .service_context("explorer_client".into());
+        let explorer_client = ExplorerHttpClient::new(config, context.log().clone())?;
+        explorer_client.auto_update(
+            context.executor,
+            vec![ProcessType::Beacon, 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/config.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 explorer_metrics: Option<explorer_api::Config>,
     pub slasher: Option<slasher::Config>,
 }
 
@@ -87,6 +88,7 @@ impl Default for Config {
             graffiti: Graffiti::default(),
             http_api: <_>::default(),
             http_metrics: <_>::default(),
+            explorer_metrics: None,
             slasher: None,
             validator_monitor_auto: false,
             validator_monitor_pubkeys: vec![],

beacon_node/client/src/metrics.rs

@@ -6,4 +6,9 @@ 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"
+    );
 }

beacon_node/client/src/notifier.rs

@@ -123,6 +123,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 +152,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 +169,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/eth1/src/metrics.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", "Check if eth1 node is connected"
+    );
+
+    pub static ref ETH1_FALLBACK_CONFIGURED: Result<IntGauge> = try_create_int_gauge(
+        "sync_eth1_fallback_configured", "Number of configured eth1 fallbacks"
+    );
+
+    // TODO(pawan): how to check fallbacks connection status? Should we refactor to enable this
+    pub static ref ETH1_FALLBACK_CONNECTED: Result<IntGauge> = try_create_int_gauge(
+        "eth1_sync_fallback_connected", "Check if eth1 fallback node is configured"
+    );
+
+
 }

beacon_node/eth1/src/service.rs

@@ -85,6 +85,9 @@ impl EndpointsCache {
         *value = Some(state);
         if state.is_err() {
             crate::metrics::inc_counter_vec(&crate::metrics::ENDPOINT_ERRORS, &[&endpoint.0]);
+            crate::metrics::set_gauge(&metrics::ETH1_CONNECTED, 0);
+        } else {
+            crate::metrics::set_gauge(&metrics::ETH1_CONNECTED, 1);
         }
         state
     }
@@ -720,13 +723,16 @@ 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);
         handle.spawn(update_future, "eth1");
     }
 

beacon_node/http_metrics/src/explorer_metrics.rs

@@ -0,0 +1,67 @@
+use crate::Context;
+use beacon_chain::BeaconChainTypes;
+use lighthouse_metrics::{json_encoder::JsonEncoder, Encoder};
+
+pub use lighthouse_metrics::*;
+
+/// Process names which need to be encoded
+/// Note: Only Gauge and Counter metrics can be encoded.
+pub const BEACON_PROCESS_METRICS: &[&str] = &[
+    "cpu_process_seconds_total",
+    "process_virtual_memory_bytes",
+    "sync_eth1_fallback_configured",
+    "sync_eth1_connected",
+    "store_disk_db_size",
+    "libp2p_peer_connected_peers_total",
+    "beacon_head_state_slot",
+    "sync_eth2_synced",
+];
+
+pub fn gather_required_metrics<T: BeaconChainTypes>(
+    ctx: &Context<T>,
+) -> std::result::Result<String, String> {
+    let mut buffer = vec![];
+    let encoder = JsonEncoder::new(
+        BEACON_PROCESS_METRICS
+            .iter()
+            .map(|m| m.to_string())
+            .collect(),
+    );
+
+    // There are two categories of metrics:
+    //
+    // - Dynamically updated: things like histograms and event counters that are updated on the
+    // fly.
+    // - Statically updated: things which are only updated at the time of the scrape (used where we
+    // can avoid cluttering up code with metrics calls).
+    //
+    // The `lighthouse_metrics` crate has a `DEFAULT_REGISTRY` global singleton (via `lazy_static`)
+    // which keeps the state of all the metrics. Dynamically updated things will already be
+    // up-to-date in the registry (because they update themselves) however statically updated
+    // things need to be "scraped".
+    //
+    // We proceed by, first updating all the static metrics using `scrape_for_metrics(..)`. Then,
+    // using `lighthouse_metrics::gather(..)` to collect the global `DEFAULT_REGISTRY` metrics into
+    // a string that can be returned via HTTP.
+
+    if let Some(beacon_chain) = ctx.chain.as_ref() {
+        slot_clock::scrape_for_metrics::<T::EthSpec, T::SlotClock>(&beacon_chain.slot_clock);
+        beacon_chain::scrape_for_metrics(beacon_chain);
+    }
+
+    if let (Some(db_path), Some(freezer_db_path)) =
+        (ctx.db_path.as_ref(), ctx.freezer_db_path.as_ref())
+    {
+        store::scrape_for_metrics(db_path, freezer_db_path);
+    }
+
+    eth2_libp2p::scrape_discovery_metrics();
+
+    warp_utils::metrics::scrape_process_health_metrics();
+
+    let metrics = lighthouse_metrics::gather();
+
+    encoder.encode(&metrics, &mut buffer).unwrap();
+
+    String::from_utf8(buffer).map_err(|e| format!("Failed to encode prometheus info: {:?}", e))
+}

beacon_node/http_metrics/src/lib.rs

@@ -1,6 +1,7 @@
 //! This crate provides a HTTP server that is solely dedicated to serving the `/metrics` endpoint.
 //!
 //! For other endpoints, see the `http_api` crate.
+mod explorer_metrics;
 mod metrics;
 
 use beacon_chain::{BeaconChain, BeaconChainTypes};
@@ -106,7 +107,8 @@ pub fn serve<T: BeaconChainTypes>(
     }
 
     let inner_ctx = ctx.clone();
-    let routes = warp::get()
+    let inner_ctx1 = ctx.clone();
+    let metrics = warp::get()
         .and(warp::path("metrics"))
         .map(move || inner_ctx.clone())
         .and_then(|ctx: Arc<Context<T>>| async move {
@@ -121,7 +123,27 @@ pub fn serve<T: BeaconChainTypes>(
                             .unwrap()
                     }),
             )
-        })
+        });
+
+    let beacon_process_metrics = warp::get()
+        .and(warp::path("beacon_process"))
+        .and(warp::path::end())
+        .map(move || inner_ctx1.clone())
+        .and_then(|ctx: Arc<Context<T>>| async move {
+            Ok::<_, warp::Rejection>(
+                explorer_metrics::gather_required_metrics(&ctx)
+                    .map(|body| Response::builder().status(200).body(body).unwrap())
+                    .unwrap_or_else(|e| {
+                        Response::builder()
+                            .status(500)
+                            .header("Content-Type", "text/plain")
+                            .body(format!("Unable to gather metrics: {:?}", e))
+                            .unwrap()
+                    }),
+            )
+        });
+    let routes = warp::any()
+        .and(warp::get().and(metrics.or(beacon_process_metrics)))
         // Add a `Server` header.
         .map(|reply| warp::reply::with_header(reply, "Server", &version_with_platform()))
         .with(cors_builder.build());

beacon_node/src/cli.rs

@@ -233,6 +233,27 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .takes_value(true),
         )
 
+        /*
+         * Explorer metrics
+         */
+
+         .arg(
+            Arg::with_name("explorer-metrics")
+                .long("explorer-metrics")
+                .help("Enable the monitoring service for sending system metrics to a remote endpoint. \
+                        This can be used to monitor your setup on services like the beaconcha.in api \
+                        Should enable metrics to get all process data")
+                .takes_value(false)
+                .requires_all(&["metrics", "explorer-address"]),
+        )
+        .arg(
+            Arg::with_name("explorer-address")
+                .long("explorer-address")
+                .value_name("ADDRESS")
+                .help("Set the address where the beacon node metrics would be sent")
+                .takes_value(true),
+        )
+
         /*
          * Standard staking flags
          */

beacon_node/src/config.rs

@@ -143,6 +143,24 @@ pub fn get_config<E: EthSpec>(
         client_config.http_metrics.allow_origin = Some(allow_origin.to_string());
     }
 
+    /*
+     * Explorer metrics
+     */
+    if cli_args.is_present("explorer-metrics") {
+        let explorer_endpoint = cli_args
+            .value_of("explorer-address")
+            .expect("guaranteed by clap")
+            .to_string();
+        client_config.explorer_metrics = Some(explorer_api::Config {
+            beacon_endpoint: Some(format!(
+                "http://{}:{}",
+                client_config.http_metrics.listen_addr, client_config.http_metrics.listen_port
+            )),
+            validator_endpoint: None,
+            explorer_endpoint,
+        });
+    }
+
     // Log a warning indicating an open HTTP server if it wasn't specified explicitly
     // (e.g. using the --staking flag).
     if cli_args.is_present("staking") {

beacon_node/src/lib.rs

@@ -82,6 +82,12 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
             builder
         };
 
+        let builder = if let Some(explorer_config) = &client_config.explorer_metrics {
+            builder.explorer_client(explorer_config)?
+        } else {
+            builder
+        };
+
         let builder = builder
             .beacon_chain_builder(client_genesis, client_config_1)
             .await?;