diff --git a/explorer/src/main.rs b/explorer/src/main.rs index 1fdfb9e234..0d11cbba1f 100644 --- a/explorer/src/main.rs +++ b/explorer/src/main.rs @@ -51,12 +51,12 @@ async fn main() { .to_owned(); let public_path = std::env::current_exe() - .expect("Failed to evaluate current exe path") - .parent() - .expect("the binary itself has no parent path?!") - .join("public"); + .expect("Failed to evaluate current exe path") + .parent() + .expect("the binary itself has no parent path?!") + .join("public"); - tokio::task::spawn_blocking(|| { + std::thread::spawn(|| { rocket::ignite() .mount("/", StaticFiles::from(public_path)) .launch() diff --git a/explorer/src/websockets/client.rs b/explorer/src/websockets/client.rs index b258afc6bc..4eeb8663a0 100644 --- a/explorer/src/websockets/client.rs +++ b/explorer/src/websockets/client.rs @@ -2,18 +2,24 @@ use log::*; use tokio::net::TcpStream; use tokio::stream::StreamExt; use tokio::sync::broadcast; +use tokio::time::{delay_for, Duration}; use tokio_native_tls::TlsStream; use tokio_tungstenite::tungstenite::{Error as WsError, Message}; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::{connect_async, stream::Stream}; pub(crate) type WsItem = Result; +const MAX_RECONNECTION_ATTEMPTS: u32 = 10; +const RECONNECTION_BACKOFF: Duration = Duration::from_secs(10); /// A websocket client which subscribes to the metrics centrally collected by the metrics server. /// All metrics messages get copied out to this dashboard instance's clients. pub(crate) struct MetricsWebsocketClient { + metrics_address: String, metrics_upstream: WebSocketStream>>, broadcaster: broadcast::Sender, + + reconnection_attempt: u32, } impl MetricsWebsocketClient { @@ -30,33 +36,68 @@ impl MetricsWebsocketClient { info!("Subscribed to metrics websocket at {}", metrics_address); Ok(MetricsWebsocketClient { + metrics_address: metrics_address.into(), metrics_upstream: ws_stream, broadcaster, + reconnection_attempt: 0, }) } - /// When the metrics server sends a message, it should be copied out to the server and distributed - /// to all connected clients. - fn on_message(&self, item: WsItem) { - let ws_message = match item { - Ok(message) => message, + async fn attempt_reconnection(&mut self) { + info!("attempting reconnection to metrics websocket..."); + if self.reconnection_attempt >= MAX_RECONNECTION_ATTEMPTS { + // kill the process and reset everything when service restarts + error!("failed to re-establish websocket connection to metrics server"); + std::process::exit(1) + } + + // use linear backoff to try to reconnect asap + delay_for(RECONNECTION_BACKOFF * self.reconnection_attempt).await; + + let ws_stream = match connect_async(&self.metrics_address).await { + Ok((ws_stream, _)) => ws_stream, Err(err) => { - error!("failed to obtain valid websocket message - {}", err); + self.reconnection_attempt += 1; + info!("reconnection failed... - {}", err); return; } }; + info!("reconnected!"); + + self.reconnection_attempt = 0; + self.metrics_upstream = ws_stream; + } + + /// When the metrics server sends a message, it should be copied out to the server and distributed + /// to all connected clients. + fn on_message(&self, item: WsItem) -> Result<(), WsError> { + let ws_message = match item { + Ok(message) => message, + Err(err) => { + error!("failed to obtain valid websocket message - {}", err); + return Err(err); + } + }; + match self.broadcaster.send(ws_message) { - Ok(received) => info!("broadcasted websocket metrics data to {} clients", received), - Err(_) => info!("no clients are currently subscribed"), + Ok(received) => debug!("broadcasted websocket metrics data to {} clients", received), + Err(_) => debug!("no clients are currently subscribed"), } + + Ok(()) } pub(crate) async fn run(&mut self) { - while let Some(incoming) = self.metrics_upstream.next().await { - self.on_message(incoming) + loop { + if let Some(incoming) = self.metrics_upstream.next().await { + if let Err(_) = self.on_message(incoming) { + self.attempt_reconnection().await; + } + } else { + self.attempt_reconnection().await; + } } - info!("Our metrics server subscriber is finished!") } }