adapter: Extract class for showing readyset status

This commit extracts the logic for obtaining the information for the
reuslt of a `show readyset status` call to a struct
ReadySetStatusReporter, and separates the formatting logic to a new
ReadySetStatus struct. This better encapsulates this logic and will
allow for easier re-use for a separate http/json formatting request.

Change-Id: I27caba35eda3ed8ee4a37a309cd5780957b10ffd
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/6399
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <[email protected]>

Author: Luke Osborne

readyset-adapter/src/backend.rs

@@ -112,6 +112,7 @@ use crate::metrics_handle::{MetricsHandle, MetricsSummary};
 use crate::query_handler::SetBehavior;
 use crate::query_status_cache::QueryStatusCache;
 use crate::rewrite::ProcessedQueryParams;
+use crate::status_reporter::ReadySetStatusReporter;
 pub use crate::upstream_database::UpstreamPrepare;
 use crate::utils::create_dummy_column;
 use crate::{create_dummy_schema, rewrite, QueryHandler, UpstreamDatabase, UpstreamDestination};
@@ -2157,27 +2158,13 @@ where
                 self.show_caches(query_id).await
             }
             SqlQuery::Show(ShowStatement::ReadySetStatus) => {
-                // Add upstream connectivity status
-                let mut additional_meta = if let Some(upstream) = &mut self.upstream {
-                    let connection_status = upstream
-                        .is_connected()
-                        .await
-                        .is_ok()
-                        .then(|| "Connected".to_string())
-                        .unwrap_or_else(|| "Unreachable".to_string());
-                    vec![("Database Connection".to_string(), connection_status)]
-                } else {
-                    vec![]
-                };
-                let conn_count = match &self.connections {
-                    Some(s) => s.len(),
-                    None => 0,
+                let reporter = ReadySetStatusReporter {
+                    upstream: &mut self.upstream,
+                    connector: &mut self.noria,
+                    connections: &self.connections,
+                    authority: &self.authority,
                 };
-                additional_meta.push(("Connection Count".to_string(), conn_count.to_string()));
-
-                self.noria
-                    .readyset_status(&self.authority, additional_meta)
-                    .await
+                Ok(reporter.report_status().await.into_query_result())
             }
             SqlQuery::Show(ShowStatement::ReadySetStatusAdapter) => self.readyset_adapter_status(),
             SqlQuery::Show(ShowStatement::ReadySetMigrationStatus(id)) => {

readyset-adapter/src/backend/noria_connector.rs

@@ -9,16 +9,16 @@ use nom_sql::{
     self, ColumnConstraint, DeleteStatement, DialectDisplay, Expr, InsertStatement, Relation,
     SqlIdentifier, SqlQuery, UnaryOperator, UpdateStatement,
 };
-use readyset_client::consensus::{Authority, AuthorityControl};
 use readyset_client::consistency::Timestamp;
 use readyset_client::internal::LocalNodeIndex;
 use readyset_client::recipe::changelist::{Change, ChangeList, IntoChanges};
 use readyset_client::results::{ResultIterator, Results};
+use readyset_client::status::ReadySetControllerStatus;
 use readyset_client::{
     ColumnSchema, GraphvizOptions, ReadQuery, ReaderAddress, ReaderHandle, ReadySetHandle,
     SchemaType, Table, TableOperation, View, ViewCreateRequest, ViewQuery,
 };
-use readyset_data::{DfType, DfValue, Dialect, TimestampTz};
+use readyset_data::{DfType, DfValue, Dialect};
 use readyset_errors::{
     internal_err, invariant_eq, table_err, unsupported, unsupported_err, ReadySetError,
     ReadySetResult,
@@ -854,55 +854,8 @@ impl NoriaConnector {
 
     /// Returns status provided by the Controller and persisted in the Authority. Also appends
     /// additional_meta provided by the caller to the status.
-    pub(crate) async fn readyset_status(
-        &mut self,
-        authority: &Authority,
-        mut additional_meta: Vec<(String, String)>,
-    ) -> ReadySetResult<QueryResult<'static>> {
-        let mut status =
-            match noria_await!(self.inner.get_mut()?, self.inner.get_mut()?.noria.status()) {
-                Ok(s) => <Vec<(String, String)>>::from(s),
-                Err(_) => vec![(
-                    "ReadySet Controller Status".to_string(),
-                    "Unavailable".to_string(),
-                )],
-            };
-
-        // Helper function for formatting
-        fn time_or_null(time_ms: Option<u64>) -> String {
-            if let Some(t) = time_ms {
-                TimestampTz::from_unix_ms(t).to_string()
-            } else {
-                "NULL".to_string()
-            }
-        }
-
-        if let Ok(Some(stats)) = authority.persistent_stats().await {
-            status.push((
-                "Last started Controller".to_string(),
-                time_or_null(stats.last_controller_startup),
-            ));
-            status.push((
-                "Last completed snapshot".to_string(),
-                time_or_null(stats.last_completed_snapshot),
-            ));
-            status.push((
-                "Last started replication".to_string(),
-                time_or_null(stats.last_started_replication),
-            ));
-            if let Some(err) = stats.last_replicator_error {
-                status.push(("Last replicator error".to_string(), err))
-            }
-        }
-
-        additional_meta.append(&mut status);
-
-        Ok(QueryResult::MetaVariables(
-            additional_meta
-                .into_iter()
-                .map(MetaVariable::from)
-                .collect(),
-        ))
+    pub(crate) async fn readyset_status(&mut self) -> ReadySetResult<ReadySetControllerStatus> {
+        noria_await!(self.inner.get_mut()?, self.inner.get_mut()?.noria.status())
     }
 
     /// Query the status of a pending migration identified by the given `migration_id`. Once the

readyset-adapter/src/lib.rs

@@ -14,6 +14,7 @@ pub mod proxied_queries_reporter;
 mod query_handler;
 pub mod query_status_cache;
 pub mod rewrite;
+mod status_reporter;
 pub mod upstream_database;
 mod utils;
 pub mod views_synchronizer;

readyset-adapter/src/status_reporter.rs

@@ -0,0 +1,111 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use crossbeam_skiplist::SkipSet;
+use readyset_client::consensus::{Authority, AuthorityControl};
+use readyset_client::debug::stats::PersistentStats;
+use readyset_client::status::ReadySetControllerStatus;
+use readyset_data::TimestampTz;
+
+use crate::backend::noria_connector::{MetaVariable, QueryResult};
+use crate::backend::NoriaConnector;
+use crate::UpstreamDatabase;
+
+#[derive(Debug)]
+pub(crate) struct ReadySetStatus {
+    pub(crate) controller_status: Option<ReadySetControllerStatus>,
+    pub(crate) connected: bool,
+    pub(crate) connection_count: usize,
+    pub(crate) persistent_stats: Option<PersistentStats>,
+}
+
+impl ReadySetStatus {
+    pub(crate) fn into_query_result(self) -> QueryResult<'static> {
+        let mut status = Vec::new();
+        status.push((
+            "Database Connection".to_string(),
+            if self.connected {
+                "Connected".to_string()
+            } else {
+                "Unreachable".to_string()
+            },
+        ));
+
+        status.push((
+            "Connection Count".to_string(),
+            self.connection_count.to_string(),
+        ));
+
+        if let Some(controller_status) = self.controller_status {
+            status.append(&mut Vec::<(String, String)>::from(controller_status));
+        } else {
+            status.push((
+                "ReadySet Controller Status".to_string(),
+                "Unavailable".to_string(),
+            ));
+        }
+
+        if let Some(stats) = self.persistent_stats {
+            status.push((
+                "Last started Controller".to_string(),
+                time_or_null(stats.last_controller_startup),
+            ));
+            status.push((
+                "Last completed snapshot".to_string(),
+                time_or_null(stats.last_completed_snapshot),
+            ));
+            status.push((
+                "Last started replication".to_string(),
+                time_or_null(stats.last_started_replication),
+            ));
+            if let Some(err) = stats.last_replicator_error {
+                status.push(("Last replicator error".to_string(), err))
+            }
+        }
+
+        QueryResult::MetaVariables(status.into_iter().map(MetaVariable::from).collect())
+    }
+}
+
+pub(crate) struct ReadySetStatusReporter<'a, DB>
+where
+    DB: UpstreamDatabase,
+{
+    pub(crate) upstream: &'a mut Option<DB>,
+    pub(crate) connector: &'a mut NoriaConnector,
+    pub(crate) connections: &'a Option<Arc<SkipSet<SocketAddr>>>,
+    pub(crate) authority: &'a Arc<Authority>,
+}
+
+impl<DB: UpstreamDatabase> ReadySetStatusReporter<'_, DB> {
+    pub(crate) async fn report_status(self) -> ReadySetStatus {
+        let controller_status = self.connector.readyset_status().await.ok();
+        let connected = if let Some(upstream) = self.upstream.as_mut() {
+            upstream.is_connected().await.unwrap_or_default()
+        } else {
+            false
+        };
+        let connection_count = self
+            .connections
+            .as_ref()
+            .map(|c| c.len())
+            .unwrap_or_default();
+        let persistent_stats = self.authority.persistent_stats().await.unwrap_or_default();
+
+        ReadySetStatus {
+            controller_status,
+            connected,
+            connection_count,
+            persistent_stats,
+        }
+    }
+}
+
+// Helper function for formatting
+fn time_or_null(time_ms: Option<u64>) -> String {
+    if let Some(t) = time_ms {
+        TimestampTz::from_unix_ms(t).to_string()
+    } else {
+        "NULL".to_string()
+    }
+}

readyset-client/src/debug/stats.rs

@@ -45,7 +45,7 @@ pub struct NodeStats {
 }
 
 /// Status that we persist in the Authority to make it available across restarts.
-#[derive(Clone, Default, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct PersistentStats {
     /// Time in millis when the controller last started up.
     pub last_controller_startup: Option<u64>,

readyset-mysql/tests/integration.rs

@@ -1834,7 +1834,10 @@ async fn show_readyset_status() {
 
     // NOTE: If this readyset extension has changed, verify the new behavior is correct then update
     // the expected values below
-    assert_eq!(ret.len(), 5);
+    assert_eq!(ret.len(), 6);
+    let row = ret.remove(0);
+    assert_eq!(row.get::<String, _>(0).unwrap(), "Database Connection");
+    assert_eq!(row.get::<String, _>(1).unwrap(), "Unreachable");
     let row = ret.remove(0);
     assert_eq!(row.get::<String, _>(0).unwrap(), "Connection Count");
     assert_eq!(row.get::<String, _>(1).unwrap(), "0");