Logging via the HTTP API (sigp#4074)

This PR adds the ability to read the Lighthouse logs from the HTTP API for both the BN and the VC.

This is done in such a way to as minimize any kind of performance hit by adding this feature.

The current design creates a tokio broadcast channel and mixes is into a form of slog drain that combines with our main global logger drain, only if the http api is enabled.

The drain gets the logs, checks the log level and drops them if they are below INFO. If they are INFO or higher, it sends them via a broadcast channel only if there are users subscribed to the HTTP API channel. If not, it drops the logs.

If there are more than one subscriber, the channel clones the log records and converts them to json in their independent HTTP API tasks.

Co-authored-by: Michael Sproul <[email protected]>

Author: 2 people

beacon_node/client/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/src/builder.rs

@@ -486,6 +486,7 @@ where
                         network_globals: None,
                         eth1_service: Some(genesis_service.eth1_service.clone()),
                         log: context.log().clone(),
+                        sse_logging_components: runtime_context.sse_logging_components.clone(),
                     });
 
                     // Discard the error from the oneshot.
@@ -706,6 +707,7 @@ where
                 network_senders: self.network_senders.clone(),
                 network_globals: self.network_globals.clone(),
                 eth1_service: self.eth1_service.clone(),
+                sse_logging_components: runtime_context.sse_logging_components.clone(),
                 log: log.clone(),
             });
 

beacon_node/http_api/Cargo.toml

@@ -36,11 +36,11 @@ tree_hash = "0.5.0"
 sysinfo = "0.26.5"
 system_health = { path = "../../common/system_health" }
 directory = { path = "../../common/directory" }
+logging = { path = "../../common/logging" }
 ethereum_serde_utils = "0.5.0"
 operation_pool = { path = "../operation_pool" }
 sensitive_url = { path = "../../common/sensitive_url" }
 unused_port = {path = "../../common/unused_port"}
-logging = { path = "../../common/logging" }
 store = { path = "../store" }
 
 [dev-dependencies]
@@ -51,4 +51,4 @@ genesis = { path = "../genesis" }
 
 [[test]]
 name = "bn_http_api_tests"
-path = "tests/main.rs"
+path = "tests/main.rs"

beacon_node/http_api/src/lib.rs

@@ -36,6 +36,7 @@ use eth2::types::{
 };
 use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
 use lighthouse_version::version_with_platform;
+use logging::SSELoggingComponents;
 use network::{NetworkMessage, NetworkSenders, ValidatorSubscriptionMessage};
 use operation_pool::ReceivedPreCapella;
 use parking_lot::RwLock;
@@ -108,6 +109,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,
 }
 
@@ -448,6 +450,9 @@ 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()));
     {
@@ -3760,6 +3765,44 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // Subscribe to logs via Server Side Events
+    // /lighthouse/logs
+    let lighthouse_log_events = warp::path("lighthouse")
+        .and(warp::path("logs"))
+        .and(warp::path::end())
+        .and(sse_component_filter)
+        .and_then(|sse_component: Option<SSELoggingComponents>| {
+            blocking_response_task(move || {
+                if let Some(logging_components) = sse_component {
+                    // Build a JSON stream
+                    let s =
+                        BroadcastStream::new(logging_components.sender.subscribe()).map(|msg| {
+                            match msg {
+                                Ok(data) => {
+                                    // Serialize to json
+                                    match data.to_json_string() {
+                                        // Send the json as a Server Side Event
+                                        Ok(json) => Ok(Event::default().data(json)),
+                                        Err(e) => Err(warp_utils::reject::server_sent_event_error(
+                                            format!("Unable to serialize to JSON {}", e),
+                                        )),
+                                    }
+                                }
+                                Err(e) => Err(warp_utils::reject::server_sent_event_error(
+                                    format!("Unable to receive event {}", e),
+                                )),
+                            }
+                        });
+
+                    Ok::<_, warp::Rejection>(warp::sse::reply(warp::sse::keep_alive().stream(s)))
+                } else {
+                    Err(warp_utils::reject::custom_server_error(
+                        "SSE Logging is not enabled".to_string(),
+                    ))
+                }
+            })
+        });
+
     // Define the ultimate set of routes that will be provided to the server.
     // Use `uor` rather than `or` in order to simplify types (see `UnifyingOrFilter`).
     let routes = warp::get()
@@ -3828,6 +3871,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .uor(get_lighthouse_merge_readiness)
                 .uor(get_lighthouse_blobs_sidecars.boxed())
                 .uor(get_events)
+                .uor(lighthouse_log_events.boxed())
                 .recover(warp_utils::reject::handle_rejection),
         )
         .boxed()

beacon_node/http_api/src/test_utils.rs

@@ -195,6 +195,7 @@ pub async fn create_api_server_on_port<T: BeaconChainTypes>(
         network_senders: Some(network_senders),
         network_globals: Some(network_globals),
         eth1_service: Some(eth1_service),
+        sse_logging_components: None,
         log,
     });
 

beacon_node/src/cli.rs

@@ -1123,7 +1123,7 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .long("gui")
                 .hidden(true)
                 .help("Enable the graphical user interface and all its requirements. \
-                      This is equivalent to --http and --validator-monitor-auto.")
+                      This enables --http and --validator-monitor-auto and enables SSE logging.")
                 .takes_value(false)
         )
         .arg(

book/src/api-lighthouse.md

@@ -679,3 +679,31 @@ 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. It is only enabled when the `--gui` flag is set in the CLI.
+
+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-vc-endpoints.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  | None                                       |
+| 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
+  }
+}
+```

common/logging/Cargo.toml

@@ -10,6 +10,13 @@ test_logger = [] # Print log output to stderr when running tests instead of drop
 [dependencies]
 slog = "2.5.2"
 slog-term = "2.6.0"
+tokio = { version = "1.26.0", features = ["sync"] }
 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"
+parking_lot = "0.12.1"
+serde = "1.0.153" 
+serde_json = "1.0.94"
+chrono = "0.4.23"