A working version

Author: AgeManning

Cargo.lock

@@ -35,7 +35,6 @@ 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

@@ -19,7 +19,6 @@ 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};
@@ -73,7 +72,6 @@ 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>,
 
 }
 
@@ -89,7 +87,7 @@ where
     /// Instantiates a new, empty builder.
     ///
     /// The `eth_spec_instance` parameter is used to concretize `TEthSpec`.
-    pub fn new(eth_spec_instance: TEthSpec, sse_logging_components: Option<SSELoggingComponents>) -> Self {
+    pub fn new(eth_spec_instance: TEthSpec) -> Self {
         Self {
             slot_clock: None,
             store: None,
@@ -107,7 +105,6 @@ where
             http_metrics_config: <_>::default(),
             slasher: None,
             eth_spec_instance,
-            sse_logging_components,
         }
     }
 
@@ -476,7 +473,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(),
+                        sse_logging_components: runtime_context.sse_logging_components.clone(),
                     });
 
                     // Discard the error from the oneshot.
@@ -601,12 +598,6 @@ 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;
@@ -703,7 +694,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(),
+                sse_logging_components: runtime_context.sse_logging_components.clone(),
                 log: log.clone(),
             });
 

beacon_node/client/src/builder.rs

@@ -21,7 +21,6 @@ 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> =
@@ -46,10 +45,9 @@ 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, sse_log).await
+        Self::new(context, client_config).await
     }
 
     /// Starts a new beacon node `Client` in the given `environment`.
@@ -58,7 +56,6 @@ 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();
@@ -83,7 +80,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
             TimeoutRwLock::disable_timeouts()
         }
 
-        let builder = ClientBuilder::new(context.eth_spec_instance.clone(), sse_logging_components)
+        let builder = ClientBuilder::new(context.eth_spec_instance.clone())
             .runtime_context(context)
             .chain_spec(spec)
             .http_api_config(client_config.http_api.clone())

beacon_node/src/lib.rs

@@ -677,3 +677,32 @@ Caveats:
   This is because the state _prior_ to the `start_epoch` needs to be loaded from the database, and
   loading a state on a boundary is most efficient.
 
+
+### `/lighthouse/logs`
+
+This is a Server Side Event subscription endpoint. This allows a user to read
+the Lighthouse logs directly from the HTTP API endpoint. This currently
+exposes INFO and higher level logs.
+
+Example:
+
+```bash
+curl -N "http://localhost:5052/lighthouse/logs"
+```
+
+Should provide an output that emits log events as they occur:
+```json
+{ 
+"data": {
+	  "time": "Mar 13 15:28:41",
+	  "level": "INFO",
+	  "msg": "Syncing",
+	  "service": "slot_notifier",
+	  "est_time": "1 hr 27 mins",
+	  "speed": "5.33 slots/sec",
+	  "distance": "28141 slots (3 days 21 hrs)",
+	  "peers": "8"
+	}
+}
+```
+

book/src/api-lighthouse.md

@@ -578,3 +578,33 @@ The following fields may be omitted or nullified to obtain default values:
 ### Example Response Body
 
 *No data is included in the response body.*
+
+## `GET /lighthouse/logs`
+
+Provides a subscription to receive logs as Server Side Events. Currently the
+logs emitted are INFO level or higher.
+
+### HTTP Specification
+
+| Property          | Specification                              |
+|-------------------|--------------------------------------------|
+| Path              | `/lighthouse/logs`                    |
+| Method            | GET                                        |
+| Required Headers  | [`Authorization`](./api-vc-auth-header.md) |
+| Typical Responses | 200                                        |
+
+### Example Response Body
+
+```json
+{
+  "data": {
+	  "time": "Mar 13 15:26:53",
+	  "level": "INFO",
+	  "msg": "Connected to beacon node(s)",
+	  "service": "notifier",
+	  "synced": 1,
+	  "available": 1,
+	  "total": 1
+  }
+}
+```

book/src/api-vc-endpoints.md

@@ -19,3 +19,4 @@ take_mut = "0.2.2"
 parking_lot = "0.12.1"
 serde = "1.0.153" 
 serde_json = "1.0.94"
+chrono = "0.4.23"

common/logging/Cargo.toml

@@ -158,17 +158,27 @@ impl Serialize for AsyncRecord {
     where
         S: serde::Serializer,
     {
+        // Get the current time
+        let dt = chrono::Local::now().format("%b %e %T").to_string();
+
         let rs = RecordStatic {
             location: &*self.location,
             level: self.level,
             tag: &self.tag,
         };
         let mut map_serializer = SerdeSerializer::new(serializer)?;
+
+        // Serialize the time and log level first
+        map_serializer.serialize_entry("time", &dt)?;
+        map_serializer.serialize_entry("level", &self.level.as_short_str())?;
+
         let kv = self.kv.lock();
 
         // Convoluted pattern to avoid binding `format_args!` to a temporary.
         // See: https://stackoverflow.com/questions/56304313/cannot-use-format-args-due-to-temporary-value-is-freed-at-the-end-of-this-state
         let mut f = |msg: std::fmt::Arguments| {
+            map_serializer.serialize_entry("msg", &msg.to_string())?;
+
             let record = Record::new(&rs, &msg, BorrowedKV(&(*kv)));
             self.logger_values
                 .serialize(&record, &mut map_serializer)
@@ -179,7 +189,6 @@ impl Serialize for AsyncRecord {
                 .map_err(serde::ser::Error::custom)
         };
         f(format_args!("{}", self.msg))?;
-
         map_serializer.end()
     }
 }
@@ -195,6 +204,14 @@ impl<S: serde::Serializer> SerdeSerializer<S> {
         Ok(SerdeSerializer { ser_map })
     }
 
+    fn serialize_entry<K, V>(&mut self, key: K, value: V) -> Result<(), S::Error>
+    where
+        K: serde::Serialize,
+        V: serde::Serialize,
+    {
+        self.ser_map.serialize_entry(&key, &value)
+    }
+
     /// Finish serialization, and return the serializer
     fn end(self) -> Result<S::Ok, S::Error> {
         self.ser_map.end()

common/logging/src/async_record.rs

@@ -7,6 +7,11 @@ use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
 use tokio::sync::broadcast::Sender;
 
+/// Default log level for SSE Events.
+// NOTE: Made this a constant. Debug level seems to be pretty intense. Can make this
+// configurable later if needed.
+const LOG_LEVEL: slog::Level = slog::Level::Info;
+
 /// The components required in the HTTP API task to receive logged events.
 #[derive(Clone)]
 pub struct SSELoggingComponents {
@@ -29,10 +34,12 @@ impl Drain for SSELoggingComponents {
     type Err = &'static str;
 
     fn log(&self, record: &Record, logger_values: &OwnedKVList) -> Result<Self::Ok, Self::Err> {
-        // There are subscribers, attempt to send the logs
-        match self.sender.send(AsyncRecord::from(record, logger_values)) {
-            Ok(_num_sent) => {} // Everything got sent
-            Err(_err) => {}     // There are no subscribers, do nothing
+        if record.level().is_at_least(LOG_LEVEL) {
+            // There are subscribers, attempt to send the logs
+            match self.sender.send(AsyncRecord::from(record, logger_values)) {
+                Ok(_num_sent) => {} // Everything got sent
+                Err(_err) => {}     // There are no subscribers, do nothing
+            }
         }
         Ok(())
     }

common/logging/src/sse_logging_components.rs

@@ -90,6 +90,7 @@ pub struct RuntimeContext<E: EthSpec> {
     pub eth_spec_instance: E,
     pub eth2_config: Eth2Config,
     pub eth2_network_config: Option<Arc<Eth2NetworkConfig>>,
+    pub sse_logging_components: Option<SSELoggingComponents>,
 }
 
 impl<E: EthSpec> RuntimeContext<E> {
@@ -102,6 +103,7 @@ impl<E: EthSpec> RuntimeContext<E> {
             eth_spec_instance: self.eth_spec_instance.clone(),
             eth2_config: self.eth2_config.clone(),
             eth2_network_config: self.eth2_network_config.clone(),
+            sse_logging_components: self.sse_logging_components.clone(),
         }
     }
 
@@ -325,6 +327,7 @@ impl<E: EthSpec> EnvironmentBuilder<E> {
 
             log = Logger::root(Duplicate::new(log, sse_logger).fuse(), o!());
         }
+
         self.log = Some(log);
 
         Ok(self)
@@ -414,6 +417,7 @@ impl<E: EthSpec> Environment<E> {
             eth_spec_instance: self.eth_spec_instance.clone(),
             eth2_config: self.eth2_config.clone(),
             eth2_network_config: self.eth2_network_config.clone(),
+            sse_logging_components: self.sse_logging_components.clone(),
         }
     }
 
@@ -429,6 +433,7 @@ impl<E: EthSpec> Environment<E> {
             eth_spec_instance: self.eth_spec_instance.clone(),
             eth2_config: self.eth2_config.clone(),
             eth2_network_config: self.eth2_network_config.clone(),
+            sse_logging_components: self.sse_logging_components.clone(),
         }
     }
 
@@ -550,13 +555,6 @@ impl<E: EthSpec> Environment<E> {
         }
     }
 
-    /// Takes the receiver of the SSE Log components if one exists.
-    /// This is used to obtain the receiver of the SSE log channel in order to pass it to the HTTP
-    /// API.
-    pub fn get_sse_log(&mut self) -> Option<SSELoggingComponents> {
-        self.sse_logging_components.take()
-    }
-
     /// Fire exit signal which shuts down all spawned services
     pub fn fire_signal(&mut self) {
         if let Some(signal) = self.signal.take() {