implement Nexus quiesce (sagas, db activity) for upgrade (#8740)

Author: davepacheco

Cargo.lock

@@ -5,7 +5,10 @@
 //! Interface for making API requests to the Oxide control plane at large
 //! from within the control plane
 
+use iddqd::IdOrdItem;
+use iddqd::id_upcast;
 use std::collections::HashMap;
+use uuid::Uuid;
 
 progenitor::generate_api!(
     spec = "../../openapi/nexus-internal.json",
@@ -85,6 +88,26 @@ progenitor::generate_api!(
     }
 );
 
+impl IdOrdItem for types::RunningSagaInfo {
+    type Key<'a> = Uuid;
+
+    fn key(&self) -> Self::Key<'_> {
+        self.saga_id
+    }
+
+    id_upcast!();
+}
+
+impl IdOrdItem for types::HeldDbClaimInfo {
+    type Key<'a> = u64;
+
+    fn key(&self) -> Self::Key<'_> {
+        self.id
+    }
+
+    id_upcast!();
+}
+
 impl omicron_common::api::external::ClientError for types::Error {
     fn message(&self) -> String {
         self.message.clone()

clients/nexus-client/src/lib.rs

@@ -5,6 +5,7 @@
 //! omdb commands that query or update specific Nexus instances
 
 mod chicken_switches;
+mod quiesce;
 mod update_status;
 
 use crate::Omdb;
@@ -79,6 +80,8 @@ use omicron_uuid_kinds::ParseError;
 use omicron_uuid_kinds::PhysicalDiskUuid;
 use omicron_uuid_kinds::SledUuid;
 use omicron_uuid_kinds::SupportBundleUuid;
+use quiesce::QuiesceArgs;
+use quiesce::cmd_nexus_quiesce;
 use serde::Deserialize;
 use slog_error_chain::InlineErrorChain;
 use std::collections::BTreeMap;
@@ -138,6 +141,8 @@ enum NexusCommands {
     MgsUpdates,
     /// interact with oximeter read policy
     OximeterReadPolicy(OximeterReadPolicyArgs),
+    /// view or modify the quiesce status
+    Quiesce(QuiesceArgs),
     /// view sagas, create and complete demo sagas
     Sagas(SagasArgs),
     /// interact with sleds
@@ -718,6 +723,10 @@ impl NexusArgs {
                 }
             },
 
+            NexusCommands::Quiesce(args) => {
+                cmd_nexus_quiesce(&omdb, &client, args).await
+            }
+
             NexusCommands::Sagas(SagasArgs { command }) => {
                 if self.nexus_internal_url.is_none() {
                     eprintln!(

dev-tools/omdb/src/bin/omdb/nexus.rs

@@ -0,0 +1,168 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+//! omdb commands for managing Nexus quiesce state
+
+use crate::Omdb;
+use crate::check_allow_destructive::DestructiveOperationToken;
+use anyhow::Context;
+use chrono::TimeDelta;
+use chrono::Utc;
+use clap::Args;
+use clap::Subcommand;
+use nexus_client::types::QuiesceState;
+use std::time::Duration;
+
+#[derive(Debug, Args)]
+pub struct QuiesceArgs {
+    #[command(subcommand)]
+    command: QuiesceCommands,
+}
+
+#[derive(Debug, Subcommand)]
+pub enum QuiesceCommands {
+    /// Show the current Nexus quiesce status
+    Show(QuiesceShowArgs),
+
+    /// Start quiescing Nexus
+    Start,
+}
+
+#[derive(Debug, Args)]
+pub struct QuiesceShowArgs {
+    /// Show details about held database connections
+    #[clap(short, long, default_value_t = false)]
+    verbose: bool,
+}
+
+pub async fn cmd_nexus_quiesce(
+    omdb: &Omdb,
+    client: &nexus_client::Client,
+    args: &QuiesceArgs,
+) -> Result<(), anyhow::Error> {
+    match &args.command {
+        QuiesceCommands::Show(args) => quiesce_show(&client, args).await,
+        QuiesceCommands::Start => {
+            let token = omdb.check_allow_destructive()?;
+            quiesce_start(&client, token).await
+        }
+    }
+}
+
+async fn quiesce_show(
+    client: &nexus_client::Client,
+    args: &QuiesceShowArgs,
+) -> Result<(), anyhow::Error> {
+    let now = Utc::now();
+    let quiesce = client
+        .quiesce_get()
+        .await
+        .context("fetching quiesce state")?
+        .into_inner();
+    match quiesce.state {
+        QuiesceState::Running => {
+            println!("running normally (not quiesced, not quiescing)");
+        }
+        QuiesceState::WaitingForSagas { time_requested } => {
+            println!(
+                "quiescing since {} ({} ago)",
+                humantime::format_rfc3339_millis(time_requested.into()),
+                format_time_delta(now - time_requested),
+            );
+            println!("details: waiting for running sagas to finish");
+        }
+        QuiesceState::WaitingForDb {
+            time_requested,
+            duration_waiting_for_sagas,
+            ..
+        } => {
+            println!(
+                "quiescing since {} ({} ago)",
+                humantime::format_rfc3339_millis(time_requested.into()),
+                format_time_delta(now - time_requested),
+            );
+            println!(
+                "details: waiting for database connections to be released"
+            );
+            println!(
+                "    previously: waiting for sagas took {}",
+                format_duration_ms(duration_waiting_for_sagas.into()),
+            );
+        }
+        QuiesceState::Quiesced {
+            time_quiesced,
+            duration_waiting_for_sagas,
+            duration_waiting_for_db,
+            duration_total,
+            ..
+        } => {
+            println!(
+                "quiesced since {} ({} ago)",
+                humantime::format_rfc3339_millis(time_quiesced.into()),
+                format_time_delta(now - time_quiesced),
+            );
+            println!(
+                "    waiting for sagas took {}",
+                format_duration_ms(duration_waiting_for_sagas.into()),
+            );
+            println!(
+                "    waiting for db quiesce took {}",
+                format_duration_ms(duration_waiting_for_db.into()),
+            );
+            println!(
+                "    total quiesce time: {}",
+                format_duration_ms(duration_total.into()),
+            );
+        }
+    }
+
+    println!("sagas running: {}", quiesce.sagas_running.len());
+    for saga in &quiesce.sagas_running {
+        println!(
+            "    saga {} started at {} ({})",
+            saga.saga_id,
+            humantime::format_rfc3339_millis(saga.time_started.into()),
+            saga.saga_name
+        );
+    }
+
+    println!("database connections held: {}", quiesce.db_claims.len());
+    for claim in &quiesce.db_claims {
+        println!(
+            "    claim {} held since {} ({} ago)",
+            claim.id,
+            claim.held_since,
+            format_time_delta(Utc::now() - claim.held_since),
+        );
+        if args.verbose {
+            println!("    acquired by:");
+            println!("{}", textwrap::indent(&claim.debug, "        "));
+        }
+    }
+
+    Ok(())
+}
+
+async fn quiesce_start(
+    client: &nexus_client::Client,
+    _token: DestructiveOperationToken,
+) -> Result<(), anyhow::Error> {
+    client.quiesce_start().await.context("quiescing Nexus")?;
+    quiesce_show(client, &QuiesceShowArgs { verbose: false }).await
+}
+
+fn format_duration_ms(duration: Duration) -> String {
+    // Ignore units smaller than a millisecond.
+    let elapsed = Duration::from_millis(
+        u64::try_from(duration.as_millis()).unwrap_or(u64::MAX),
+    );
+    humantime::format_duration(elapsed).to_string()
+}
+
+fn format_time_delta(time_delta: TimeDelta) -> String {
+    match time_delta.to_std() {
+        Ok(d) => format_duration_ms(d),
+        Err(_) => String::from("<time delta out of range>"),
+    }
+}

dev-tools/omdb/src/bin/omdb/nexus/quiesce.rs

@@ -849,6 +849,7 @@ Commands:
   clickhouse-policy     interact with clickhouse policy
   mgs-updates           print information about pending MGS updates
   oximeter-read-policy  interact with oximeter read policy
+  quiesce               view or modify the quiesce status
   sagas                 view sagas, create and complete demo sagas
   sleds                 interact with sleds
   support-bundles       interact with support bundles [aliases: sb]

dev-tools/omdb/tests/usage_errors.out

@@ -45,6 +45,7 @@ http.workspace = true
 http-body-util.workspace = true
 hyper.workspace = true
 hyper-staticfile.workspace = true
+iddqd.workspace = true
 id-map.workspace = true
 illumos-utils.workspace = true
 internal-dns-resolver.workspace = true

nexus/Cargo.toml

@@ -239,6 +239,50 @@ impl ApiResourceWithRolesType for Fleet {
     type AllowedRoles = FleetRole;
 }
 
+/// Represents the "quiesce" state of Nexus
+///
+/// It is essential that authorizing actions on this resource *not* access the
+/// database because we cannot do that while quiesced and we *do* want to be
+/// able to read and modify the quiesce state while quiesced.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PolarClass)]
+pub struct QuiesceState;
+/// Singleton representing the [`QuiesceState`] itself for authz purposes
+pub const QUIESCE_STATE: QuiesceState = QuiesceState;
+
+impl Eq for QuiesceState {}
+impl PartialEq for QuiesceState {
+    fn eq(&self, _: &Self) -> bool {
+        // There is only one QuiesceState
+        true
+    }
+}
+
+impl AuthorizedResource for QuiesceState {
+    fn load_roles<'fut>(
+        &'fut self,
+        _: &'fut OpContext,
+        _: &'fut authn::Context,
+        _: &'fut mut RoleSet,
+    ) -> BoxFuture<'fut, Result<(), Error>> {
+        // We don't use (database) roles to grant access to the quiesce state.
+        futures::future::ready(Ok(())).boxed()
+    }
+
+    fn on_unauthorized(
+        &self,
+        _: &Authz,
+        error: Error,
+        _: AnyActor,
+        _: Action,
+    ) -> Error {
+        error
+    }
+
+    fn polar_class(&self) -> oso::Class {
+        Self::get_polar_class()
+    }
+}
+
 // TODO: refactor synthetic resources below
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]

nexus/auth/src/authz/api_resources.rs

@@ -347,6 +347,20 @@ has_relation(fleet: Fleet, "parent_fleet", collection: SamlIdentityProvider)
 # Fleet.  None of these resources defines their own roles.
 #
 
+# Describes the quiesce state of a particular Nexus instance.
+#
+# These authz checks must not require the database.  We grant this directly to
+# callers of the internal API.
+resource QuiesceState {
+	permissions = [ "read", "modify" ];
+}
+has_permission(USER_INTERNAL_API: AuthenticatedActor, "read", _q: QuiesceState);
+has_permission(
+    USER_INTERNAL_API: AuthenticatedActor,
+    "modify",
+    _q: QuiesceState
+);
+
 # Describes the policy for reading and modifying DNS configuration
 # (both internal and external)
 resource DnsConfig {

nexus/auth/src/authz/omicron.polar

@@ -112,6 +112,7 @@ pub fn make_omicron_oso(log: &slog::Logger) -> Result<OsoInit, anyhow::Error> {
         IpPoolList::get_polar_class(),
         ConsoleSessionList::get_polar_class(),
         DeviceAuthRequestList::get_polar_class(),
+        QuiesceState::get_polar_class(),
         SiloCertificateList::get_polar_class(),
         SiloIdentityProviderList::get_polar_class(),
         SiloUserList::get_polar_class(),

nexus/auth/src/authz/oso_generic.rs

@@ -8,6 +8,7 @@ use diesel::PgConnection;
 use diesel_dtrace::DTraceConnection;
 use nexus_auth::context::OpContext;
 use omicron_common::api::external::Error;
+use std::any::Any;
 
 /// The interface between lookups and the Nexus datastore.
 #[async_trait::async_trait]
@@ -48,5 +49,40 @@ where
 // If a more natural location becomes available in the future, consider moving
 // these aliases there.
 pub type DbConnection = DTraceConnection<PgConnection>;
-pub type DataStoreConnection =
-    qorb::claim::Handle<async_bb8_diesel::Connection<DbConnection>>;
+pub type AsyncConnection = async_bb8_diesel::Connection<DbConnection>;
+
+pub struct DataStoreConnection {
+    inner: qorb::claim::Handle<AsyncConnection>,
+
+    // `DataStoreConnection` is used by various packages that we'd like to not
+    // depend on `nexus-db-queries` (in order to parallelize compilation).
+    // However, we need to do some datastore-specific work around the lifecycle
+    // of this object (i.e., when it gets instantiated and when it gets
+    // dropped).  To achieve this, the caller in `nexus-db-queries` provides a
+    // `releaser` whose sole purpose is to be dropped when this object is
+    // dropped, allowing it to do the needed cleanup there.
+    #[allow(dead_code)]
+    releaser: Box<dyn Any + Send + Sync + 'static>,
+}
+
+impl DataStoreConnection {
+    pub fn new(
+        inner: qorb::claim::Handle<AsyncConnection>,
+        releaser: Box<dyn Any + Send + Sync + 'static>,
+    ) -> DataStoreConnection {
+        DataStoreConnection { inner, releaser }
+    }
+}
+
+impl std::ops::Deref for DataStoreConnection {
+    type Target = AsyncConnection;
+    fn deref(&self) -> &Self::Target {
+        self.inner.deref()
+    }
+}
+
+impl std::ops::DerefMut for DataStoreConnection {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.inner.deref_mut()
+    }
+}