[restatectl] Improve performance and information display with unprovisioned clusters or under partial node connectivity

crates/cli-util/src/opts.rs

@@ -16,7 +16,7 @@ use cling::Collect;
 
 use restate_core::network::net_util::CommonClientConnectionOptions;
 
-const DEFAULT_CONNECT_TIMEOUT: u64 = 5_000;
+const DEFAULT_CONNECT_TIMEOUT: u64 = 3_000;
 const DEFAULT_REQUEST_TIMEOUT: u64 = 13_000;
 
 #[derive(ValueEnum, Clone, Copy, Default, PartialEq, Eq)]

tools/restatectl/Cargo.toml

@@ -44,6 +44,7 @@ cling = { workspace = true }
 crossterm = { version = "0.27.0" }
 ctrlc = { version = "3.4" }
 diff = "0.1.13"
+futures = { workspace = true }
 futures-util = { workspace = true }
 itertools = { workspace = true }
 json-patch = "2.0.0"

tools/restatectl/src/commands/log/list_logs.rs

@@ -34,7 +34,7 @@ pub async fn list_logs(connection: &ConnectionInfo, _opts: &ListLogsOpts) -> any
 
     let mut logs_table = Table::new_styled();
 
-    c_println!("Log chain {}", logs.version());
+    c_println!("Logs {}", logs.version());
 
     write_default_provider(
         &mut Console::stdout(),
@@ -45,29 +45,33 @@ pub async fn list_logs(connection: &ConnectionInfo, _opts: &ListLogsOpts) -> any
     // sort by log-id for display
     let logs: BTreeMap<LogId, &Chain> = logs.iter().map(|(id, chain)| (*id, chain)).collect();
 
-    for (log_id, chain) in logs {
-        let params = deserialize_replicated_log_params(&chain.tail());
-        logs_table.add_row(vec![
-            Cell::new(log_id),
-            Cell::new(format!("{}", &chain.tail().base_lsn)),
-            Cell::new(format!("{:?}", chain.tail().config.kind)),
-            render_loglet_params(&params, |p| Cell::new(p.loglet_id)),
-            render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.replication))),
-            render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.sequencer))),
-            render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.nodeset))),
+    if logs.is_empty() {
+        c_println!("No logs found. Has the cluster been provisioned yet?");
+    } else {
+        for (log_id, chain) in logs {
+            let params = deserialize_replicated_log_params(&chain.tail());
+            logs_table.add_row(vec![
+                Cell::new(log_id),
+                Cell::new(format!("{}", &chain.tail().base_lsn)),
+                Cell::new(format!("{:?}", chain.tail().config.kind)),
+                render_loglet_params(&params, |p| Cell::new(p.loglet_id)),
+                render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.replication))),
+                render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.sequencer))),
+                render_loglet_params(&params, |p| Cell::new(format!("{:#}", p.nodeset))),
+            ]);
+        }
+
+        logs_table.set_styled_header(vec![
+            "L-ID",
+            "FROM-LSN",
+            "KIND",
+            "LOGLET-ID",
+            "REPLICATION",
+            "SEQUENCER",
+            "NODESET",
         ]);
+        c_println!("{}", logs_table);
     }
 
-    logs_table.set_styled_header(vec![
-        "L-ID",
-        "FROM-LSN",
-        "KIND",
-        "LOGLET-ID",
-        "REPLICATION",
-        "SEQUENCER",
-        "NODESET",
-    ]);
-    c_println!("{}", logs_table);
-
     Ok(())
 }

tools/restatectl/src/commands/log/reconfigure.rs

@@ -157,7 +157,7 @@ async fn inner_reconfigure(
     c_println!(" ├ Tail LSN: {}", sealed_segment.tail_offset);
     c_println!(" ├ Provider: {}", sealed_segment.provider);
     let Ok(provider) = ProviderKind::from_str(&sealed_segment.provider, true) else {
-        c_eprintln!("Unkown provider type '{}'", sealed_segment.provider);
+        c_eprintln!("Unknown provider type '{}'", sealed_segment.provider);
         return Ok(());
     };
 
@@ -200,7 +200,7 @@ fn replicated_loglet_params(
         ReplicatedLogletParams {
             loglet_id,
             nodeset: if opts.nodeset.is_empty() {
-                anyhow::bail!("Missing nodeset. Nodeset is required if last segment is not of replicated type");
+                bail!("Missing nodeset. Nodeset is required if last segment is not of replicated type");
             } else {
                 NodeSet::from_iter(opts.nodeset.iter().cloned())
             },

tools/restatectl/src/commands/metadata_server/status.rs

@@ -10,9 +10,11 @@
 
 use std::collections::BTreeMap;
 
+use futures::future::join_all;
 use itertools::Itertools;
 use tonic::codec::CompressionEncoding;
-use tonic::IntoRequest;
+use tonic::{IntoRequest, Status};
+use tracing::debug;
 
 use restate_cli_util::_comfy_table::{Cell, Color, Table};
 use restate_cli_util::c_println;
@@ -23,9 +25,10 @@ use restate_types::protobuf::common::MetadataServerStatus;
 use restate_types::{PlainNodeId, Version};
 
 use crate::connection::ConnectionInfo;
-use crate::util::grpc_channel;
 
 pub async fn list_metadata_servers(connection: &ConnectionInfo) -> anyhow::Result<()> {
+    debug!("Gathering metadata server status information");
+
     let nodes_configuration = connection.get_nodes_configuration().await?;
     let mut metadata_nodes_table = Table::new_styled();
     let header = vec![
@@ -45,13 +48,34 @@ pub async fn list_metadata_servers(connection: &ConnectionInfo) -> anyhow::Resul
 
     let mut unreachable_nodes = BTreeMap::default();
 
-    for (node_id, node_config) in nodes_configuration.iter_role(Role::MetadataServer) {
-        let metadata_channel = grpc_channel(node_config.address.clone());
-        let mut metadata_client = MetadataServerSvcClient::new(metadata_channel)
-            .accept_compressed(CompressionEncoding::Gzip);
-
-        let metadata_store_status = metadata_client.status(().into_request()).await;
-
+    let futures =
+        nodes_configuration
+            .iter_role(Role::MetadataServer)
+            .map(|(node_id, node_config)| {
+                let address = node_config.address.clone();
+                async move {
+                    let channel = match connection.connect(&address).await {
+                        Ok(channel) => channel,
+                        Err(conn_error) => {
+                            return (node_id, Err(Status::from_error(Box::new(conn_error))));
+                        }
+                    };
+
+                    let mut metadata_client = MetadataServerSvcClient::new(channel)
+                        .accept_compressed(CompressionEncoding::Gzip);
+
+                    debug!("Querying metadata service status on node {address}");
+                    let metadata_store_status = metadata_client.status(().into_request()).await;
+
+                    (node_id, metadata_store_status)
+                }
+            });
+    let results = join_all(futures)
+        .await
+        .into_iter()
+        .collect::<BTreeMap<_, _>>();
+
+    for (node_id, metadata_store_status) in results {
         let status = match metadata_store_status {
             Ok(response) => response.into_inner(),
             Err(err) => {
@@ -119,6 +143,7 @@ pub async fn list_metadata_servers(connection: &ConnectionInfo) -> anyhow::Resul
         ]);
     }
 
+    c_println!("Metadata service");
     c_println!("{}", metadata_nodes_table);
 
     if !unreachable_nodes.is_empty() {