update quiesce states to reflect RFD 588

Author: davepacheco

nexus/reconfigurator/execution/src/lib.rs

@@ -611,50 +611,36 @@ fn register_reassign_sagas_step<'a>(
                         .into();
                 };
 
-                // Re-assign sagas, but only if we're allowed to.  If Nexus is
-                // quiescing, we don't want to assign any new sagas to
-                // ourselves.
-                let result = saga_quiesce.reassign_if_possible(async || {
-                    // For any expunged Nexus zones, re-assign in-progress sagas
-                    // to some other Nexus.  If this fails for some reason, it
-                    // doesn't affect anything else.
-                    let sec_id = nexus_db_model::SecId::from(nexus_id);
-                    let reassigned = sagas::reassign_sagas_from_expunged(
-                        opctx, datastore, blueprint, sec_id,
-                    )
-                    .await
-                    .context("failed to re-assign sagas");
-                    match reassigned {
-                        Ok(needs_saga_recovery) => (
-                            StepSuccess::new(needs_saga_recovery).build(),
-                            needs_saga_recovery,
-                        ),
-                        Err(error) => {
-                            // It's possible that we failed after having
-                            // re-assigned sagas in the database.
-                            let maybe_reassigned = true;
-                            (
-                                StepWarning::new(false, error.to_string())
-                                    .build(),
-                                maybe_reassigned,
-                            )
+                // Re-assign sagas.
+                Ok(saga_quiesce
+                    .reassign_sagas(async || {
+                        // For any expunged Nexus zones, re-assign in-progress
+                        // sagas to some other Nexus.  If this fails for some
+                        // reason, it doesn't affect anything else.
+                        let sec_id = nexus_db_model::SecId::from(nexus_id);
+                        let reassigned = sagas::reassign_sagas_from_expunged(
+                            opctx, datastore, blueprint, sec_id,
+                        )
+                        .await
+                        .context("failed to re-assign sagas");
+                        match reassigned {
+                            Ok(needs_saga_recovery) => (
+                                StepSuccess::new(needs_saga_recovery).build(),
+                                needs_saga_recovery,
+                            ),
+                            Err(error) => {
+                                // It's possible that we failed after having
+                                // re-assigned sagas in the database.
+                                let maybe_reassigned = true;
+                                (
+                                    StepWarning::new(false, error.to_string())
+                                        .build(),
+                                    maybe_reassigned,
+                                )
+                            }
                         }
-                    }
-                });
-
-                match result.await {
-                    // Re-assignment is allowed, and we did try.  It may or may
-                    // not have succeeded.  Either way, that's reflected in
-                    // `step_result`.
-                    Ok(step_result) => Ok(step_result),
-                    // Re-assignment is disallowed.  Report this step skipped
-                    // with an explanation of why.
-                    Err(error) => StepSkipped::new(
-                        false,
-                        InlineErrorChain::new(&error).to_string(),
-                    )
-                    .into(),
-                }
+                    })
+                    .await)
             },
         )
         .register()

nexus/src/app/background/init.rs

@@ -131,6 +131,7 @@ use super::tasks::vpc_routes;
 use super::tasks::webhook_deliverator;
 use crate::Nexus;
 use crate::app::oximeter::PRODUCER_LEASE_DURATION;
+use crate::app::quiesce::NexusQuiesceHandle;
 use crate::app::saga::StartSaga;
 use nexus_background_task_interface::Activator;
 use nexus_background_task_interface::BackgroundTasks;
@@ -437,7 +438,7 @@ impl BackgroundTasksInitializer {
             nexus_id,
             task_saga_recovery.clone(),
             args.mgs_updates_tx,
-            args.saga_recovery.quiesce.clone(),
+            args.nexus_quiesce,
         );
         let rx_blueprint_exec = blueprint_executor.watcher();
         driver.register(TaskDefinition {
@@ -1029,6 +1030,8 @@ pub struct BackgroundTasksData {
     pub webhook_delivery_client: reqwest::Client,
     /// Channel for configuring pending MGS updates
     pub mgs_updates_tx: watch::Sender<PendingMgsUpdates>,
+    /// handle for controlling Nexus quiesce
+    pub nexus_quiesce: NexusQuiesceHandle,
 }
 
 /// Starts the three DNS-propagation-related background tasks for either

nexus/src/app/background/tasks/blueprint_execution.rs

@@ -4,7 +4,10 @@
 
 //! Background task for realizing a plan blueprint
 
-use crate::app::background::{Activator, BackgroundTask};
+use crate::app::{
+    background::{Activator, BackgroundTask},
+    quiesce::NexusQuiesceHandle,
+};
 use futures::FutureExt;
 use futures::future::BoxFuture;
 use internal_dns_resolver::Resolver;
@@ -13,14 +16,12 @@ use nexus_db_queries::db::DataStore;
 use nexus_reconfigurator_execution::{
     RealizeBlueprintOutput, RequiredRealizeArgs,
 };
-use nexus_types::{
-    deployment::{
-        Blueprint, BlueprintTarget, PendingMgsUpdates, execution::EventBuffer,
-    },
-    quiesce::SagaQuiesceHandle,
+use nexus_types::deployment::{
+    Blueprint, BlueprintTarget, PendingMgsUpdates, execution::EventBuffer,
 };
 use omicron_uuid_kinds::OmicronZoneUuid;
 use serde_json::json;
+use slog_error_chain::InlineErrorChain;
 use std::sync::Arc;
 use tokio::sync::watch;
 use update_engine::NestedError;
@@ -35,7 +36,7 @@ pub struct BlueprintExecutor {
     tx: watch::Sender<usize>,
     saga_recovery: Activator,
     mgs_update_tx: watch::Sender<PendingMgsUpdates>,
-    saga_quiesce: SagaQuiesceHandle,
+    nexus_quiesce: NexusQuiesceHandle,
 }
 
 impl BlueprintExecutor {
@@ -48,7 +49,7 @@ impl BlueprintExecutor {
         nexus_id: OmicronZoneUuid,
         saga_recovery: Activator,
         mgs_update_tx: watch::Sender<PendingMgsUpdates>,
-        saga_quiesce: SagaQuiesceHandle,
+        nexus_quiesce: NexusQuiesceHandle,
     ) -> BlueprintExecutor {
         let (tx, _) = watch::channel(0);
         BlueprintExecutor {
@@ -59,7 +60,7 @@ impl BlueprintExecutor {
             tx,
             saga_recovery,
             mgs_update_tx,
-            saga_quiesce,
+            nexus_quiesce,
         }
     }
 
@@ -87,6 +88,47 @@ impl BlueprintExecutor {
         };
 
         let (bp_target, blueprint) = &*update;
+
+        // Regardless of anything else: propagate whatever this blueprint
+        // says about our quiescing state.
+        //
+        // During startup under normal operation, the blueprint will reflect
+        // that we're not quiescing.  Propagating this will enable sagas to
+        // be created elsewhere in Nexus.
+        //
+        // At some point during an upgrade, we'll encounter a blueprint that
+        // reflects that we are quiescing.  Propagating this will disable sagas
+        // from being created.
+        //
+        // In all other cases, this will have no effect.
+        //
+        // We do this now, before doing anything else, for two reasons: (1)
+        // during startup, we want to do this ASAP to minimize unnecessary saga
+        // creation failures (i.e., don't wait until we try to execute the
+        // blueprint before enabling sagas, since we already know if we're
+        // quiescing or not); and (2) because we want to do it even if blueprint
+        // execution is disabled.
+        match blueprint.nexus_quiescing(self.nexus_id) {
+            Ok(quiescing) => {
+                debug!(
+                    &opctx.log,
+                    "blueprint execution: quiesce check";
+                    "quiescing" => quiescing
+                );
+                self.nexus_quiesce.set_quiescing(quiescing);
+            }
+            Err(error) => {
+                // This should be impossible.  But it doesn't really affect
+                // anything else so there's no reason to stop execution.
+                error!(
+                    &opctx.log,
+                    "blueprint execution: failed to determine if this Nexus \
+                     is quiescing";
+                    InlineErrorChain::new(&*error)
+                );
+            }
+        };
+
         if !bp_target.enabled {
             warn!(&opctx.log,
                       "Blueprint execution: skipped";
@@ -119,7 +161,7 @@ impl BlueprintExecutor {
                 blueprint,
                 sender,
                 mgs_updates: self.mgs_update_tx.clone(),
-                saga_quiesce: self.saga_quiesce.clone(),
+                saga_quiesce: self.nexus_quiesce.sagas(),
             }
             .as_nexus(self.nexus_id),
         )
@@ -181,6 +223,7 @@ impl BackgroundTask for BlueprintExecutor {
 mod test {
     use super::BlueprintExecutor;
     use crate::app::background::{Activator, BackgroundTask};
+    use crate::app::quiesce::NexusQuiesceHandle;
     use httptest::Expectation;
     use httptest::matchers::{not, request};
     use httptest::responders::status_code;
@@ -207,7 +250,6 @@ mod test {
         PlanningReport, blueprint_zone_type,
     };
     use nexus_types::external_api::views::SledState;
-    use nexus_types::quiesce::SagaQuiesceHandle;
     use omicron_common::api::external;
     use omicron_common::api::external::Generation;
     use omicron_common::zpool_name::ZpoolName;
@@ -390,7 +432,7 @@ mod test {
             OmicronZoneUuid::new_v4(),
             Activator::new(),
             dummy_tx,
-            SagaQuiesceHandle::new(opctx.log.clone()),
+            NexusQuiesceHandle::new(&opctx.log, datastore.clone()),
         );
 
         // Now we're ready.

nexus/src/app/background/tasks/blueprint_planner.rs

@@ -278,18 +278,15 @@ impl BackgroundTask for BlueprintPlanner {
 #[cfg(test)]
 mod test {
     use super::*;
-    use crate::app::background::Activator;
     use crate::app::background::tasks::blueprint_execution::BlueprintExecutor;
     use crate::app::background::tasks::blueprint_load::TargetBlueprintLoader;
     use crate::app::background::tasks::inventory_collection::InventoryCollector;
+    use crate::app::{background::Activator, quiesce::NexusQuiesceHandle};
     use nexus_inventory::now_db_precision;
     use nexus_test_utils_macros::nexus_test;
-    use nexus_types::{
-        deployment::{
-            PendingMgsUpdates, PlannerChickenSwitches,
-            ReconfiguratorChickenSwitches,
-        },
-        quiesce::SagaQuiesceHandle,
+    use nexus_types::deployment::{
+        PendingMgsUpdates, PlannerChickenSwitches,
+        ReconfiguratorChickenSwitches,
     };
     use omicron_uuid_kinds::OmicronZoneUuid;
 
@@ -429,7 +426,7 @@ mod test {
             OmicronZoneUuid::new_v4(),
             Activator::new(),
             dummy_tx,
-            SagaQuiesceHandle::new(opctx.log.clone()),
+            NexusQuiesceHandle::new(&opctx.log, datastore.clone()),
         );
         let value = executor.activate(&opctx).await;
         let value = value.as_object().expect("response is not a JSON object");

nexus/src/app/mod.rs

@@ -27,7 +27,6 @@ use nexus_db_queries::db;
 use nexus_mgs_updates::ArtifactCache;
 use nexus_mgs_updates::MgsUpdateDriver;
 use nexus_types::deployment::PendingMgsUpdates;
-use nexus_types::quiesce::SagaQuiesceHandle;
 use omicron_common::address::DENDRITE_PORT;
 use omicron_common::address::MGD_PORT;
 use omicron_common::address::MGS_PORT;
@@ -111,11 +110,11 @@ pub(crate) mod sagas;
 // TODO: When referring to API types, we should try to include
 // the prefix unless it is unambiguous.
 
+use crate::app::quiesce::NexusQuiesceHandle;
 pub(crate) use nexus_db_model::MAX_NICS_PER_INSTANCE;
 pub(crate) use nexus_db_queries::db::queries::disk::MAX_DISKS_PER_INSTANCE;
 use nexus_mgs_updates::DEFAULT_RETRY_TIMEOUT;
 use nexus_types::internal_api::views::MgsUpdateDriverStatus;
-use nexus_types::internal_api::views::QuiesceState;
 use sagas::demo::CompletingDemoSagas;
 
 // XXX: Might want to recast as max *floating* IPs, we have at most one
@@ -280,11 +279,8 @@ pub struct Nexus {
     #[allow(dead_code)]
     repo_depot_resolver: Box<dyn qorb::resolver::Resolver>,
 
-    /// whether Nexus is quiescing, and how far it's gotten
-    quiesce: watch::Sender<QuiesceState>,
-
-    /// details about saga quiescing
-    saga_quiesce: SagaQuiesceHandle,
+    /// state of overall Nexus quiesce activity
+    quiesce: NexusQuiesceHandle,
 }
 
 impl Nexus {
@@ -336,6 +332,8 @@ impl Nexus {
             sec_store,
         ));
 
+        let quiesce = NexusQuiesceHandle::new(&log, db_datastore.clone());
+
         // It's a bit of a red flag to use an unbounded channel.
         //
         // This particular channel is used to send a Uuid from the saga executor
@@ -360,14 +358,11 @@ impl Nexus {
         // task.  If someone changed the config, they'd have to remember to
         // update this here.  This doesn't seem worth it.
         let (saga_create_tx, saga_recovery_rx) = mpsc::unbounded_channel();
-        let saga_quiesce = SagaQuiesceHandle::new(
-            log.new(o!("component" => "SagaQuiesceHandle")),
-        );
         let sagas = Arc::new(SagaExecutor::new(
             Arc::clone(&sec_client),
             log.new(o!("component" => "SagaExecutor")),
             saga_create_tx,
-            saga_quiesce.clone(),
+            quiesce.sagas(),
         ));
 
         // Create a channel for replicating repository artifacts. 16 is a
@@ -465,8 +460,6 @@ impl Nexus {
         let mgs_update_status_rx = mgs_update_driver.status_rx();
         let _mgs_driver_task = tokio::spawn(mgs_update_driver.run());
 
-        let (quiesce, _) = watch::channel(QuiesceState::running());
-
         let nexus = Nexus {
             id: config.deployment.id,
             rack_id,
@@ -520,7 +513,6 @@ impl Nexus {
             mgs_resolver,
             repo_depot_resolver,
             quiesce,
-            saga_quiesce,
         };
 
         // TODO-cleanup all the extra Arcs here seems wrong
@@ -570,14 +562,15 @@ impl Nexus {
                     webhook_delivery_client: task_nexus
                         .webhook_delivery_client
                         .clone(),
+                    nexus_quiesce: task_nexus.quiesce.clone(),
 
                     saga_recovery: SagaRecoveryHelpers {
                         recovery_opctx: saga_recovery_opctx,
                         maker: task_nexus.clone(),
                         sec_client: sec_client.clone(),
                         registry: sagas::ACTION_REGISTRY.clone(),
                         sagas_started_rx: saga_recovery_rx,
-                        quiesce: task_nexus.saga_quiesce.clone(),
+                        quiesce: task_nexus.quiesce.sagas(),
                     },
                     tuf_artifact_replication_rx,
                     mgs_updates_tx,