More temp commits

Author: AgeManning

Cargo.lock

@@ -35,6 +35,7 @@ serde = "1.0.116"
 clap_utils = { path = "../common/clap_utils" }
 hyper = "0.14.4"
 lighthouse_version = { path = "../common/lighthouse_version" }
+logging = { path = "../common/logging" }
 hex = "0.4.2"
 slasher = { path = "../slasher", default-features = false }
 monitoring_api = { path = "../common/monitoring_api" }

beacon_node/Cargo.toml

@@ -6,7 +6,6 @@ edition = "2021"
 
 [dev-dependencies]
 serde_yaml = "0.8.13"
-logging = { path = "../../common/logging" }
 state_processing = { path = "../../consensus/state_processing" }
 operation_pool = { path = "../operation_pool" }
 tokio = "1.14.0"
@@ -17,6 +16,7 @@ store = { path = "../store" }
 network = { path = "../network" }
 timer = { path = "../timer" }
 lighthouse_network = { path = "../lighthouse_network" }
+logging = { path = "../../common/logging" }
 parking_lot = "0.12.0"
 types = { path = "../../consensus/types" }
 eth2_config = { path = "../../common/eth2_config" }

beacon_node/client/Cargo.toml

@@ -19,6 +19,7 @@ use eth2::{
     types::{BlockId, StateId},
     BeaconNodeHttpClient, Error as ApiError, Timeouts,
 };
+use logging::SSELoggingComponents;
 use execution_layer::ExecutionLayer;
 use genesis::{interop_genesis_state, Eth1GenesisService, DEFAULT_ETH1_BLOCK_HASH};
 use lighthouse_network::{prometheus_client::registry::Registry, NetworkGlobals};
@@ -72,6 +73,8 @@ pub struct ClientBuilder<T: BeaconChainTypes> {
     http_metrics_config: http_metrics::Config,
     slasher: Option<Arc<Slasher<T::EthSpec>>>,
     eth_spec_instance: T::EthSpec,
+    sse_logging_components: Option<SSELoggingComponents>,
+
 }
 
 impl<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>
@@ -86,7 +89,7 @@ where
     /// Instantiates a new, empty builder.
     ///
     /// The `eth_spec_instance` parameter is used to concretize `TEthSpec`.
-    pub fn new(eth_spec_instance: TEthSpec) -> Self {
+    pub fn new(eth_spec_instance: TEthSpec, sse_logging_components: Option<SSELoggingComponents>) -> Self {
         Self {
             slot_clock: None,
             store: None,
@@ -104,6 +107,7 @@ where
             http_metrics_config: <_>::default(),
             slasher: None,
             eth_spec_instance,
+            sse_logging_components,
         }
     }
 
@@ -472,6 +476,7 @@ where
                         network_globals: None,
                         eth1_service: Some(genesis_service.eth1_service.clone()),
                         log: context.log().clone(),
+                        sse_logging_components: self.sse_logging_components.take(),
                     });
 
                     // Discard the error from the oneshot.
@@ -596,6 +601,12 @@ where
         self
     }
 
+    /// Adds SSE Logging components to the builder if they exist. 
+    pub fn sse_logging_comonents(mut self, components: Option<SSELoggingComponents>) -> Self {
+        self.sse_logging_components = components;
+        self
+    }
+
     /// Provides configuration for the HTTP server that serves Prometheus metrics.
     pub fn http_metrics_config(mut self, config: http_metrics::Config) -> Self {
         self.http_metrics_config = config;
@@ -692,6 +703,7 @@ where
                 network_senders: self.network_senders.clone(),
                 network_globals: self.network_globals.clone(),
                 eth1_service: self.eth1_service.clone(),
+                sse_logging_components: self.sse_logging_components.take(),
                 log: log.clone(),
             });
 

beacon_node/client/src/builder.rs

@@ -36,6 +36,7 @@ tree_hash = "0.4.1"
 sysinfo = "0.26.5"
 system_health = { path = "../../common/system_health" }
 directory = { path = "../../common/directory" }
+logging = { path = "../../common/logging" }
 eth2_serde_utils = "0.1.1"
 operation_pool = { path = "../operation_pool" }
 

beacon_node/http_api/Cargo.toml

@@ -73,6 +73,7 @@ use warp_utils::{
     query::multi_key_query,
     task::{blocking_json_task, blocking_task},
 };
+use logging::SSELoggingComponents;
 
 const API_PREFIX: &str = "eth";
 
@@ -105,6 +106,7 @@ pub struct Context<T: BeaconChainTypes> {
     pub network_senders: Option<NetworkSenders<T::EthSpec>>,
     pub network_globals: Option<Arc<NetworkGlobals<T::EthSpec>>>,
     pub eth1_service: Option<eth1::Service>,
+    pub sse_logging_components: Option<SSELoggingComponents>,
     pub log: Logger,
 }
 
@@ -445,6 +447,10 @@ pub fn serve<T: BeaconChainTypes>(
     let inner_ctx = ctx.clone();
     let log_filter = warp::any().map(move || inner_ctx.log.clone());
 
+    let inner_components = ctx.sse_logging_components.clone();
+    let sse_component_filter = warp::any().map(move || inner_components.clone());
+
+
     // Create a `warp` filter that provides access to local system information.
     let system_info = Arc::new(RwLock::new(sysinfo::System::new()));
     {
@@ -3560,6 +3566,80 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+
+    // Subscribe to logs via Server Side Events
+    // /lighthouse/logs
+    let get_events = warp::path("lighthouse") 
+        .and(warp::path("logs"))
+        .and(warp::path::end())
+        .and_then(
+            |topics_res: Result<api_types::EventQuery, warp::Rejection>,
+             chain: Arc<BeaconChain<T>>| {
+                blocking_task(move || {
+                    let topics = topics_res?;
+                    // for each topic subscribed spawn a new subscription
+                    let mut receivers = Vec::with_capacity(topics.topics.len());
+
+                    if let Some(event_handler) = chain.event_handler.as_ref() {
+                        for topic in topics.topics {
+                            let receiver = match topic {
+                                api_types::EventTopic::Head => event_handler.subscribe_head(),
+                                api_types::EventTopic::Block => event_handler.subscribe_block(),
+                                api_types::EventTopic::Attestation => {
+                                    event_handler.subscribe_attestation()
+                                }
+                                api_types::EventTopic::VoluntaryExit => {
+                                    event_handler.subscribe_exit()
+                                }
+                                api_types::EventTopic::FinalizedCheckpoint => {
+                                    event_handler.subscribe_finalized()
+                                }
+                                api_types::EventTopic::ChainReorg => {
+                                    event_handler.subscribe_reorgs()
+                                }
+                                api_types::EventTopic::ContributionAndProof => {
+                                    event_handler.subscribe_contributions()
+                                }
+                                api_types::EventTopic::LateHead => {
+                                    event_handler.subscribe_late_head()
+                                }
+                                api_types::EventTopic::BlockReward => {
+                                    event_handler.subscribe_block_reward()
+                                }
+                            };
+
+                            receivers.push(BroadcastStream::new(receiver).map(|msg| {
+                                match msg {
+                                    Ok(data) => Event::default()
+                                        .event(data.topic_name())
+                                        .json_data(data)
+                                        .map_err(|e| {
+                                            warp_utils::reject::server_sent_event_error(format!(
+                                                "{:?}",
+                                                e
+                                            ))
+                                        }),
+                                    Err(e) => Err(warp_utils::reject::server_sent_event_error(
+                                        format!("{:?}", e),
+                                    )),
+                                }
+                            }));
+                        }
+                    } else {
+                        return Err(warp_utils::reject::custom_server_error(
+                            "event handler was not initialized".to_string(),
+                        ));
+                    }
+
+                    let s = futures::stream::select_all(receivers);
+
+                    Ok::<_, warp::Rejection>(warp::sse::reply(warp::sse::keep_alive().stream(s)))
+                })
+            },
+        );
+
+
+
     // Define the ultimate set of routes that will be provided to the server.
     let routes = warp::get()
         .and(

beacon_node/http_api/src/lib.rs

@@ -21,6 +21,7 @@ use slog::{info, warn};
 use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 use types::EthSpec;
+use logging::SSELoggingComponents;
 
 /// A type-alias to the tighten the definition of a production-intended `Client`.
 pub type ProductionClient<E> =
@@ -45,9 +46,10 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
     pub async fn new_from_cli(
         context: RuntimeContext<E>,
         matches: ArgMatches<'static>,
+        sse_log: Option<SSELoggingComponents>,
     ) -> Result<Self, String> {
         let client_config = get_config::<E>(&matches, &context)?;
-        Self::new(context, client_config).await
+        Self::new(context, client_config, sse_log).await
     }
 
     /// Starts a new beacon node `Client` in the given `environment`.
@@ -56,6 +58,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
     pub async fn new(
         context: RuntimeContext<E>,
         mut client_config: ClientConfig,
+        sse_logging_components: Option<SSELoggingComponents>,
     ) -> Result<Self, String> {
         let spec = context.eth2_config().spec.clone();
         let client_genesis = client_config.genesis.clone();
@@ -80,7 +83,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
             TimeoutRwLock::disable_timeouts()
         }
 
-        let builder = ClientBuilder::new(context.eth_spec_instance.clone())
+        let builder = ClientBuilder::new(context.eth_spec_instance.clone(), sse_logging_components)
             .runtime_context(context)
             .chain_spec(spec)
             .http_api_config(client_config.http_api.clone())

beacon_node/src/lib.rs

@@ -10,6 +10,9 @@ test_logger = [] # Print log output to stderr when running tests instead of drop
 [dependencies]
 slog = "2.5.2"
 slog-term = "2.6.0"
+tokio = "1.14.0"
 lighthouse_metrics = { path = "../lighthouse_metrics" }
 lazy_static = "1.4.0"
 sloggers = { version = "2.1.1", features = ["json"] }
+slog-async = "2.7.0"
+take_mut = "0.2.2"