diff --git a/dev-tools/omdb/src/bin/omdb/nexus.rs b/dev-tools/omdb/src/bin/omdb/nexus.rs index b6f6bac7853..5ab76a3c880 100644 --- a/dev-tools/omdb/src/bin/omdb/nexus.rs +++ b/dev-tools/omdb/src/bin/omdb/nexus.rs @@ -1296,6 +1296,9 @@ fn print_task_details(bgtask: &BackgroundTask, details: &serde_json::Value) { "abandoned_vmm_reaper" => { print_task_abandoned_vmm_reaper(details); } + "vmm_stop_for_update" => { + print_task_vmm_stop_for_update(details); + } "attached_subnet_manager" => { print_task_attached_subnet_manager_status(details); } @@ -1510,6 +1513,58 @@ fn print_task_abandoned_vmm_reaper(details: &serde_json::Value) { }; } +fn print_task_vmm_stop_for_update(details: &serde_json::Value) { + use nexus_types::internal_api::background::VmmStopForUpdateStatus; + + match serde_json::from_value::(details.clone()) { + Err(error) => eprintln!( + "warning: failed to interpret task details: {:?}: {:?}", + error, details + ), + Ok(VmmStopForUpdateStatus { + vmms_stopped_by_sled, + vmms_failed_to_stop_by_sled, + error_messages, + }) => { + let total: usize = + vmms_stopped_by_sled.iter().map(|s| s.vmm_ids.len()).sum(); + println!(" VMMs stopped for sled evacuation: {total} total"); + for sled in vmms_stopped_by_sled { + println!( + " sled_id={} n_stopped={}", + sled.sled_id, + sled.vmm_ids.len() + ); + } + if !vmms_failed_to_stop_by_sled.is_empty() { + let total_failed: usize = vmms_failed_to_stop_by_sled + .iter() + .map(|s| s.vmm_ids.len()) + .sum(); + println!(" VMMs that failed to stop: {total_failed} total"); + for sled in vmms_failed_to_stop_by_sled { + println!( + " sled_id={} n_failed_to_stop={}", + sled.sled_id, + sled.vmm_ids.len() + ); + for vmm in sled.vmm_ids { + println!(" vmm_id={}", vmm); + } + } + } + + if !error_messages.is_empty() { + let total_err = error_messages.len(); + println!(" Error messages: {total_err} total"); + for error in error_messages { + println!(" {error}"); + } + } + } + }; +} + fn print_task_blueprint_planner(details: &serde_json::Value) { let status = match serde_json::from_value::(details.clone()) diff --git a/dev-tools/omdb/tests/env.out b/dev-tools/omdb/tests/env.out index 9ade7712ae8..8054218dd88 100644 --- a/dev-tools/omdb/tests/env.out +++ b/dev-tools/omdb/tests/env.out @@ -271,6 +271,10 @@ task: "vmm_mark_stop_for_update" marks VMMs on evacuating sleds as needing to be stopped for an update +task: "vmm_stop_for_update" + stops VMMs that are marked to be stopped for a sled update + + task: "vpc_route_manager" propagates updated VPC routes to all OPTE ports @@ -547,6 +551,10 @@ task: "vmm_mark_stop_for_update" marks VMMs on evacuating sleds as needing to be stopped for an update +task: "vmm_stop_for_update" + stops VMMs that are marked to be stopped for a sled update + + task: "vpc_route_manager" propagates updated VPC routes to all OPTE ports @@ -810,6 +818,10 @@ task: "vmm_mark_stop_for_update" marks VMMs on evacuating sleds as needing to be stopped for an update +task: "vmm_stop_for_update" + stops VMMs that are marked to be stopped for a sled update + + task: "vpc_route_manager" propagates updated VPC routes to all OPTE ports diff --git a/dev-tools/omdb/tests/successes.out b/dev-tools/omdb/tests/successes.out index dcbf3b3e98e..d7dcd0e5d56 100644 --- a/dev-tools/omdb/tests/successes.out +++ b/dev-tools/omdb/tests/successes.out @@ -601,6 +601,10 @@ task: "vmm_mark_stop_for_update" marks VMMs on evacuating sleds as needing to be stopped for an update +task: "vmm_stop_for_update" + stops VMMs that are marked to be stopped for a sled update + + task: "vpc_route_manager" propagates updated VPC routes to all OPTE ports @@ -1209,6 +1213,12 @@ task: "vmm_mark_stop_for_update" batch size: 1000 VMMs marked to be stopped for an update: 0 +task: "vmm_stop_for_update" + configured period: every m + last completed activation: , triggered by + started at (s ago) and ran for ms + VMMs stopped for sled evacuation: 0 total + task: "vpc_route_manager" configured period: every s last completed activation: , triggered by @@ -1946,6 +1956,12 @@ task: "vmm_mark_stop_for_update" batch size: 1000 VMMs marked to be stopped for an update: 0 +task: "vmm_stop_for_update" + configured period: every m + last completed activation: , triggered by + started at (s ago) and ran for ms + VMMs stopped for sled evacuation: 0 total + task: "vpc_route_manager" configured period: every s last completed activation: , triggered by diff --git a/nexus-config/src/nexus_config.rs b/nexus-config/src/nexus_config.rs index 9bdfbbd9e90..ef6cb2cf6bc 100644 --- a/nexus-config/src/nexus_config.rs +++ b/nexus-config/src/nexus_config.rs @@ -457,6 +457,9 @@ pub struct BackgroundTaskConfig { pub v2p_mapping_propagation: V2PMappingPropagationConfig, /// configuration for abandoned VMM reaper task pub abandoned_vmm_reaper: AbandonedVmmReaperConfig, + /// configuration for the task that stops VMMs marked to be stopped for an + /// update + pub vmm_stop_for_update: VmmStopForUpdateConfig, /// configuration for saga recovery task pub saga_recovery: SagaRecoveryConfig, /// configuration for lookup region port task @@ -825,6 +828,14 @@ pub struct AbandonedVmmReaperConfig { pub period_secs: Duration, } +#[serde_as] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct VmmStopForUpdateConfig { + /// period (in seconds) for periodic activations of this background task + #[serde_as(as = "DurationSeconds")] + pub period_secs: Duration, +} + #[serde_as] #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct SagaRecoveryConfig { @@ -1387,6 +1398,7 @@ mod test { service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 + vmm_stop_for_update.period_secs = 300 saga_recovery.period_secs = 60 lookup_region_port.period_secs = 60 region_snapshot_replacement_start.period_secs = 30 @@ -1627,6 +1639,9 @@ mod test { abandoned_vmm_reaper: AbandonedVmmReaperConfig { period_secs: Duration::from_secs(60), }, + vmm_stop_for_update: VmmStopForUpdateConfig { + period_secs: Duration::from_secs(300), + }, saga_recovery: SagaRecoveryConfig { period_secs: Duration::from_secs(60), }, @@ -1803,6 +1818,7 @@ mod test { service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 + vmm_stop_for_update.period_secs = 300 saga_recovery.period_secs = 60 lookup_region_port.period_secs = 60 region_snapshot_replacement_start.period_secs = 30 diff --git a/nexus/background-task-interface/src/init.rs b/nexus/background-task-interface/src/init.rs index b71458ca410..69b05400d69 100644 --- a/nexus/background-task-interface/src/init.rs +++ b/nexus/background-task-interface/src/init.rs @@ -36,6 +36,7 @@ pub struct BackgroundTasks { pub task_instance_reincarnation: Activator, pub task_service_firewall_propagation: Activator, pub task_abandoned_vmm_reaper: Activator, + pub task_vmm_stop_for_update: Activator, pub task_audit_log_cleanup: Activator, pub task_audit_log_timeout_incomplete: Activator, pub task_vpc_route_manager: Activator, diff --git a/nexus/db-model/src/vmm_state.rs b/nexus/db-model/src/vmm_state.rs index 51718113920..fa978c6a92a 100644 --- a/nexus/db-model/src/vmm_state.rs +++ b/nexus/db-model/src/vmm_state.rs @@ -107,16 +107,29 @@ impl VmmState { self.to_nexus_state().exists_on_sled() } + /// The states in which a VMM should be marked to be stopped during sled + /// evacuation. + pub const SHOULD_MARK_TO_STOP_FOR_EVACUATION: &'static [Self] = &[ + // A VMM in one of these states is on its way, or is already running, + // and can be marked to be stopped. + VmmState::Creating, + VmmState::Starting, + VmmState::Running, + VmmState::Rebooting, + // If it is not in one of these states, it is already stopping/stopped, + // migrating, or has terminated, so it does not need to be marked. + ]; + /// The states in which a VMM should be stopped during sled evacuation. pub const SHOULD_STOP_FOR_EVACUATION: &'static [Self] = &[ // A VMM in one of these states is on its way, or is already running, - // and can be stopped. - VmmState::Creating, + // is registered with a sled agent, and can be stopped. VmmState::Starting, VmmState::Running, VmmState::Rebooting, // If it is not in one of these states, it is already stopping/stopped, - // migrating, or has terminated, so it does not need to be stopped. + // migrating, is not registered with a sled agent yet, or has + // terminated, so it does not need to be stopped. ]; } @@ -451,7 +464,7 @@ mod tests { } #[test] - fn test_should_stop_for_update_states_consistent() { + fn test_should_mark_to_stop_for_update_states_consistent() { // This test exists to make sure that when a new state is added, we // consider whether it is a state where reconfigurator should mark the // VMM to be stopped for sled evacuation. More detailed information about @@ -469,12 +482,52 @@ mod tests { VmmState::Creating | VmmState::Starting | VmmState::Running + | VmmState::Rebooting => { + assert!( + VmmState::SHOULD_MARK_TO_STOP_FOR_EVACUATION + .contains(state) + ); + } + VmmState::Failed + | VmmState::Stopping + | VmmState::Stopped + | VmmState::Migrating + | VmmState::Destroyed + | VmmState::SagaUnwound => { + assert!( + !VmmState::SHOULD_MARK_TO_STOP_FOR_EVACUATION + .contains(state) + ); + } + } + } + } + + #[test] + fn test_should_stop_for_update_states_consistent() { + // This test exists to make sure that when a new state is added, we + // consider whether it is a state where reconfigurator should stop a VMM + // for sled evacuation. More detailed information about the process of + // restarting instances during a live update can be found in RFD 739. + // + // The use of an exhaustive `match state` here ensures that the addition + // of a new `VmmState` variant will result in a compiler error until it + // is added to this test. If you have added a new variant to VmmState, + // consider whether or not that state should be added to the list of + // states in which VMMs are stopped. Do *not* change the match + // here to be non-exhaustive! + for state in VmmState::ALL_STATES.iter() { + match state { + VmmState::Starting + | VmmState::Running | VmmState::Rebooting => { assert!( VmmState::SHOULD_STOP_FOR_EVACUATION.contains(state) ); } + VmmState::Failed + | VmmState::Creating | VmmState::Stopping | VmmState::Stopped | VmmState::Migrating diff --git a/nexus/db-queries/src/db/datastore/vmm.rs b/nexus/db-queries/src/db/datastore/vmm.rs index c7377f5e5bd..8efe06e8a8e 100644 --- a/nexus/db-queries/src/db/datastore/vmm.rs +++ b/nexus/db-queries/src/db/datastore/vmm.rs @@ -11,6 +11,7 @@ use crate::db::model; use crate::db::model::Vmm; use crate::db::model::VmmState as DbVmmState; use crate::db::pagination::paginated; +use crate::db::pagination::paginated_multicolumn; use crate::db::update_and_check::UpdateAndCheck; use crate::db::update_and_check::UpdateAndQueryResult; use crate::db::update_and_check::UpdateStatus; @@ -24,10 +25,12 @@ use nexus_db_errors::ErrorHandler; use nexus_db_errors::OptionalError; use nexus_db_errors::public_error_from_diesel; use nexus_db_lookup::DbConnection; +use nexus_db_model::ApplySledFilterExt; use nexus_db_schema::enums::SledBpAvailabilityEnum; use nexus_db_schema::enums::VmmStateEnum; use nexus_db_schema::schema::vmm; use nexus_db_schema::schema::vmm::dsl; +use nexus_types::deployment::SledFilter; use nexus_types::instance::Migrations; use nexus_types::instance::VmmRuntimeState; use omicron_common::api::external::CreateResult; @@ -175,9 +178,9 @@ impl DataStore { /// Marks VMMs sleds that are evacuating as needing to be stopped in order /// to update the sled. /// - /// VMMs that are not in one of the `SHOULD_STOP_FOR_EVACUATION` states do - /// not need to be stopped, so they are left untouched. VMMs that are - /// already marked to be stopped by update are also excluded. + /// VMMs that are not in one of the `SHOULD_MARK_TO_STOP_FOR_EVACUATION` + /// states do not need to be stopped, so they are left untouched. VMMs that + /// are already marked to be stopped by update are also excluded. /// /// The task to stop instances only needs to know whether a VMM should be /// stopped or not, so a boolean would be enough. But, we mark the VMMs @@ -215,7 +218,7 @@ impl DataStore { model::DbSledBpAvailability::Unavailable, ) .bind::, _>( - DbVmmState::SHOULD_STOP_FOR_EVACUATION.to_vec(), + DbVmmState::SHOULD_MARK_TO_STOP_FOR_EVACUATION.to_vec(), ) .bind::(i64::from(batch_size.get())) .execute_async(&*conn) @@ -569,17 +572,44 @@ impl DataStore { .await .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server)) } + + /// Lists VMMs that have been marked as needing to be stopped for a sled + /// update, and that *can* be stopped. + /// + /// The query performed by this function is paginated by the sled and + /// VMM UUIDs, in that order. + pub async fn vmm_list_stoppable_for_update_by_sled_id( + &self, + opctx: &OpContext, + pagparams: &DataPageParams<'_, (Uuid, Uuid)>, + ) -> ListResultVec { + use nexus_db_schema::schema::sled::dsl as sled_dsl; + + paginated_multicolumn(dsl::vmm, (dsl::sled_id, dsl::id), pagparams) + .inner_join(sled_dsl::sled.on(sled_dsl::id.eq(dsl::sled_id))) + .filter(dsl::time_deleted.is_null()) + .filter(dsl::stop_for_update_disposition_generation.is_not_null()) + .filter(dsl::state.eq_any(DbVmmState::SHOULD_STOP_FOR_EVACUATION)) + .sled_filter(SledFilter::InService) + .select(Vmm::as_select()) + .load_async(&*self.pool_connection_authorized(opctx).await?) + .await + .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server)) + } } #[cfg(test)] mod tests { use super::*; + use crate::authz; use crate::db; use crate::db::model::Generation; use crate::db::model::Migration; use crate::db::pub_test_utils::TestDatabase; + use crate::db::pub_test_utils::helpers::SledUpdateBuilder; use crate::db::pub_test_utils::helpers::create_vmm_for_instance; use iddqd::IdOrdMap; + use nexus_db_lookup::LookupPath; use nexus_db_model::ActiveSledBpAvailability; use nexus_db_model::SledBlueprintAvailabilityInput; use nexus_db_model::SledBpAvailabilityState; @@ -592,6 +622,7 @@ mod tests { use omicron_uuid_kinds::InstanceUuid; use omicron_uuid_kinds::SledUuid; use sled_agent_types::instance::MigrationState; + use std::collections::BTreeSet; use std::collections::HashMap; #[tokio::test] @@ -1325,6 +1356,130 @@ mod tests { logctx.cleanup_successful(); } + #[tokio::test] + async fn test_vmm_list_stoppable_for_update_by_sled_id() { + // Setup + let logctx = dev::test_setup_log( + "test_vmm_list_stoppable_for_update_by_sled_id", + ); + let db = TestDatabase::new_with_datastore(&logctx.log).await; + let (opctx, datastore) = (db.opctx(), db.datastore()); + let conn = datastore.pool_connection_for_tests().await.unwrap(); + + // The states from which a VMM can still be stopped for an update. Only + // marked VMMs in these states should be listed by + // `vmm_list_stoppable_for_update_by_sled_id`. + let is_stoppable = |state: DbVmmState| { + matches!( + state, + DbVmmState::Starting + | DbVmmState::Running + | DbVmmState::Rebooting + ) + }; + + // Insert 2 VMMs (one marked to stop, one not) in each state per sled + // (2 in-service and 1 expunged). + let generation = UpdateDispositionGeneration::from(1); + let sled_a = SledUuid::new_v4(); + let sled_b = SledUuid::new_v4(); + let sled_expunged = SledUuid::new_v4(); + let mut vmms = Vec::new(); + for sled_id in [sled_a, sled_b, sled_expunged] { + // The query only returns VMMs on in-service sleds. We store the + // three sleds in the db. + datastore + .sled_upsert(SledUpdateBuilder::new().sled_id(sled_id).build()) + .await + .expect("sled should be upserted successfully"); + + // New sleds are in-service by default. We expunge one of them to + // test that no VMMs from that sled are returned. + if sled_id == sled_expunged { + let (.., authz_sled) = LookupPath::new(&opctx, datastore) + .sled_id(sled_id) + .lookup_for(authz::Action::Modify) + .await + .expect("sled should exist"); + datastore + .sled_set_policy_to_expunged(&opctx, &authz_sled) + .await + .expect("sled should be expunged"); + } + + for &state in DbVmmState::ALL_STATES { + for marker in [Some(generation), None] { + let failure_reason = (state == DbVmmState::Failed) + .then_some(db::model::VmmFailureReason::FromSledAgent); + let vmm = datastore + .vmm_insert( + &opctx, + Vmm { + id: Uuid::new_v4(), + time_created: Utc::now(), + time_deleted: None, + instance_id: Uuid::new_v4(), + sled_id: sled_id.into(), + propolis_ip: "10.1.9.32".parse().unwrap(), + propolis_port: 420.into(), + cpu_platform: VmmCpuPlatform::SledDefault, + time_state_updated: Utc::now(), + generation: Generation::new(), + state, + failure_reason, + stop_for_update_disposition_generation: marker + .map(Into::into), + }, + ) + .await + .expect("VMM should be inserted successfully"); + vmms.push(vmm); + } + } + } + + // There should be 60 VMM rows. 2 VMMs (one marked, one not) per state + // per sled + let row_count: i64 = dsl::vmm + // Add a filter to ensure there is no full table scan + .filter(dsl::id.ne(Uuid::nil())) + .count() + .get_result_async(&*conn) + .await + .expect("VMMs should be counted"); + assert_eq!(row_count, 60); + + // Check there are 6 stoppable VMMs (3 per in-service sled). + let expected_ids: BTreeSet = vmms + .iter() + .filter(|vmm| { + vmm.sled_id() != sled_expunged + && vmm.stop_for_update_disposition_generation.is_some() + && is_stoppable(vmm.state) + }) + .map(|vmm| vmm.id) + .collect(); + assert_eq!(expected_ids.len(), 6); + + // Check the retrieved ids for stoppable VMMs are the same as the + // expected ones + let listed = datastore + .vmm_list_stoppable_for_update_by_sled_id( + &opctx, + &DataPageParams::max_page(), + ) + .await + .expect("marked VMMs should be listed"); + assert_eq!(listed.len(), expected_ids.len()); + let listed_ids: BTreeSet = + listed.iter().map(|vmm| vmm.id).collect(); + assert_eq!(listed_ids, expected_ids); + + // Clean up. + db.terminate().await; + logctx.cleanup_successful(); + } + async fn mark_failed_no_such_instance( opctx: &OpContext, datastore: &DataStore, diff --git a/nexus/examples/config-second.toml b/nexus/examples/config-second.toml index baa812dc75d..9f16dc2fc06 100644 --- a/nexus/examples/config-second.toml +++ b/nexus/examples/config-second.toml @@ -162,6 +162,7 @@ instance_reincarnation.period_secs = 60 service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 +vmm_stop_for_update.period_secs = 60 saga_recovery.period_secs = 600 lookup_region_port.period_secs = 60 region_snapshot_replacement_start.period_secs = 30 diff --git a/nexus/examples/config.toml b/nexus/examples/config.toml index 7fe5e6f5b6e..17bdf0ffc8b 100644 --- a/nexus/examples/config.toml +++ b/nexus/examples/config.toml @@ -153,6 +153,7 @@ instance_reincarnation.period_secs = 60 service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 +vmm_stop_for_update.period_secs = 60 saga_recovery.period_secs = 600 lookup_region_port.period_secs = 60 region_snapshot_replacement_start.period_secs = 30 diff --git a/nexus/src/app/background/init.rs b/nexus/src/app/background/init.rs index 45c82899d39..26a44e4fbd4 100644 --- a/nexus/src/app/background/init.rs +++ b/nexus/src/app/background/init.rs @@ -148,6 +148,7 @@ use super::tasks::vpc_routes; use super::tasks::webhook_deliverator; use crate::Nexus; use crate::app::background::tasks::populate_switch_ports; +use crate::app::background::tasks::vmm_stop_for_update; use crate::app::external_client::ExternalHttpClient; use crate::app::oximeter::PRODUCER_LEASE_DURATION; use crate::app::quiesce::NexusQuiesceHandle; @@ -254,6 +255,7 @@ impl BackgroundTasksInitializer { task_instance_reincarnation: Activator::new(), task_service_firewall_propagation: Activator::new(), task_abandoned_vmm_reaper: Activator::new(), + task_vmm_stop_for_update: Activator::new(), task_audit_log_cleanup: Activator::new(), task_audit_log_timeout_incomplete: Activator::new(), task_vpc_route_manager: Activator::new(), @@ -353,6 +355,7 @@ impl BackgroundTasksInitializer { task_instance_reincarnation, task_service_firewall_propagation, task_abandoned_vmm_reaper, + task_vmm_stop_for_update, task_vpc_route_manager, task_saga_recovery, task_lookup_region_port, @@ -1341,12 +1344,16 @@ impl BackgroundTasksInitializer { activator: task_populate_switch_ports, }); + let vmm_mark_stop_for_update = + VmmMarkStopForUpdate::new(datastore.clone()); + let vmm_mark_stop_for_update_watcher = + vmm_mark_stop_for_update.watcher(); driver.register(TaskDefinition { name: "vmm_mark_stop_for_update", description: "marks VMMs on evacuating sleds as needing to be \ stopped for an update", period: config.vmm_mark_stop_for_update.period_secs, - task_impl: Box::new(VmmMarkStopForUpdate::new(datastore)), + task_impl: Box::new(vmm_mark_stop_for_update), opctx: opctx.child(BTreeMap::new()), // We activate this task any time the blueprint_rendezvous task runs. // We want to err on the side of running this task more often than @@ -1356,6 +1363,23 @@ impl BackgroundTasksInitializer { activator: task_vmm_mark_stop_for_update, }); + driver.register(TaskDefinition { + name: "vmm_stop_for_update", + description: "stops VMMs that are marked to be stopped for a sled \ + update", + period: config.vmm_stop_for_update.period_secs, + task_impl: Box::new(vmm_stop_for_update::VmmStopForUpdate::new( + datastore, + sagas.clone(), + )), + opctx: opctx.child(BTreeMap::new()), + // Like the vmm_mark_stop_for_update task, this task will be + // activated each time that task runs regardless of the outcome. + // We do this in the spirit of futureproofing. + watchers: vec![Box::new(vmm_mark_stop_for_update_watcher)], + activator: task_vmm_stop_for_update, + }); + driver } } diff --git a/nexus/src/app/background/tasks/mod.rs b/nexus/src/app/background/tasks/mod.rs index aabf38770a3..4d24722d16e 100644 --- a/nexus/src/app/background/tasks/mod.rs +++ b/nexus/src/app/background/tasks/mod.rs @@ -58,5 +58,6 @@ pub mod tuf_artifact_replication; pub mod tuf_repo_pruner; pub mod v2p_mappings; pub mod vmm_mark_stop_for_update; +pub mod vmm_stop_for_update; pub mod vpc_routes; pub mod webhook_deliverator; diff --git a/nexus/src/app/background/tasks/vmm_mark_stop_for_update.rs b/nexus/src/app/background/tasks/vmm_mark_stop_for_update.rs index 02469495bfc..9e186d55c6f 100644 --- a/nexus/src/app/background/tasks/vmm_mark_stop_for_update.rs +++ b/nexus/src/app/background/tasks/vmm_mark_stop_for_update.rs @@ -21,14 +21,21 @@ use nexus_types::internal_api::background::VmmMarkStopForUpdateStatus; use serde_json::json; use slog_error_chain::InlineErrorChain; use std::sync::Arc; +use tokio::sync::watch; pub struct VmmMarkStopForUpdate { datastore: Arc, + tx: watch::Sender, } impl VmmMarkStopForUpdate { pub fn new(datastore: Arc) -> Self { - Self { datastore } + let (tx, _rx) = watch::channel(0); + Self { datastore, tx } + } + + pub fn watcher(&self) -> watch::Receiver { + self.tx.subscribe() } pub(crate) async fn actually_activate( @@ -68,6 +75,9 @@ impl VmmMarkStopForUpdate { ); } + // Trigger anything waiting on this task to run. + self.tx.send_modify(|count| *count = *count + 1); + VmmMarkStopForUpdateStatus { vmms_marked, batches, diff --git a/nexus/src/app/background/tasks/vmm_stop_for_update.rs b/nexus/src/app/background/tasks/vmm_stop_for_update.rs new file mode 100644 index 00000000000..0f020aa530f --- /dev/null +++ b/nexus/src/app/background/tasks/vmm_stop_for_update.rs @@ -0,0 +1,1125 @@ +// 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/. + +//! Background task that stops VMMs that have been marked as needing to be +//! stopped for a sled update. + +use crate::app::background::BackgroundTask; +use crate::app::instance; +use crate::app::instance::SledAgentInstanceError; +use crate::app::saga::StartSaga; +use futures::future::BoxFuture; +use iddqd::IdOrdMap; +use nexus_db_model::Vmm; +use nexus_db_queries::context::OpContext; +use nexus_db_queries::db::DataStore; +use nexus_db_queries::db::datastore::SQL_BATCH_SIZE; +use nexus_db_queries::db::pagination::Paginator; +use nexus_types::instance::SledVmmState; +use nexus_types::instance::VmmFailureReason; +use nexus_types::instance::VmmRuntimeState; +use nexus_types::instance::VmmState; +use nexus_types::internal_api::background::VmmStopForUpdateStatus; +use nexus_types::internal_api::background::VmmsBySled; +use omicron_common::api::external::Error; +use omicron_uuid_kinds::GenericUuid; +use omicron_uuid_kinds::InstanceUuid; +use omicron_uuid_kinds::PropolisUuid; +use omicron_uuid_kinds::SledUuid; +use parallel_task_set::ParallelTaskSet; +use reqwest::Client as ReqwestClient; +use serde_json::json; +use sled_agent_client::types::VmmPutStateBody; +use sled_agent_client::types::VmmStateRequested; +use slog_error_chain::InlineErrorChain; +use std::collections::BTreeMap; +use std::collections::BTreeSet; +use std::sync::Arc; + +/// Determines the number of attempts to stop VMMs (sled agent PUT request and +/// subsequent update saga per VMM) that can execute concurrently. +/// +/// Like the instance-watcher task, we keep this number low to make sure we +/// don't starve the system of resources. +const MAX_CONCURRENT_STOPS: usize = 16; + +struct VmmStopForUpdateResult { + vmms_stopped_by_sled: IdOrdMap, + vmms_failed_to_stop_by_sled: IdOrdMap, + errors: Vec, +} + +impl VmmStopForUpdateResult { + fn new() -> Self { + Self { + vmms_stopped_by_sled: IdOrdMap::new(), + vmms_failed_to_stop_by_sled: IdOrdMap::new(), + errors: Vec::new(), + } + } + + /// Merges the values of `self` with a given [`VmmStopForUpdateResult`] + fn merge(&mut self, other: VmmStopForUpdateResult) { + let VmmStopForUpdateResult { + vmms_stopped_by_sled, + vmms_failed_to_stop_by_sled, + errors, + } = other; + + for VmmsBySled { sled_id, vmm_ids } in vmms_stopped_by_sled { + self.vmms_stopped_by_sled + .entry(sled_id) + .or_insert_with(|| VmmsBySled { + sled_id, + vmm_ids: BTreeSet::new(), + }) + .vmm_ids + .extend(vmm_ids); + } + for VmmsBySled { sled_id, vmm_ids } in vmms_failed_to_stop_by_sled { + self.vmms_failed_to_stop_by_sled + .entry(sled_id) + .or_insert_with(|| VmmsBySled { + sled_id, + vmm_ids: BTreeSet::new(), + }) + .vmm_ids + .extend(vmm_ids); + } + self.errors.extend(errors); + } +} + +pub struct VmmStopForUpdate { + datastore: Arc, + sagas: Arc, + reqwest_client: Option, +} + +impl VmmStopForUpdate { + pub(crate) fn new( + datastore: Arc, + sagas: Arc, + ) -> Self { + Self { datastore, sagas, reqwest_client: None } + } + + /// Retrieves a list of VMMs marked to be stopped for update by sled, and + /// stops them in batches. + async fn stop_all(&mut self, opctx: &OpContext) -> VmmStopForUpdateResult { + let mut result = VmmStopForUpdateResult::new(); + + // Build the reqwest client beforehand to avoid building a new one per + // page or sled below. + let reqwest_client = match self.reqwest_client.clone() { + Some(client) => client, + None => { + match nexus_networking::default_reqwest_client_builder().build() + { + Ok(client) => { + self.reqwest_client = Some(client.clone()); + client + } + Err(e) => { + let err = InlineErrorChain::new(&e); + + slog::error!( + opctx.log, + "Failed to build the HTTP client"; + &err, + ); + + result.errors.push(Error::InternalError { + internal_message: format!( + "failed to build the HTTP client: {}", + err + ), + }); + + return result; + } + } + } + }; + + let mut tasks = + ParallelTaskSet::new_with_parallelism(MAX_CONCURRENT_STOPS); + let task_opctx = Arc::new(opctx.child(BTreeMap::new())); + + let mut paginator = Paginator::new( + SQL_BATCH_SIZE, + dropshot::PaginationOrder::Ascending, + ); + while let Some(p) = paginator.next() { + let vmms = match self + .datastore + .vmm_list_stoppable_for_update_by_sled_id( + opctx, + &p.current_pagparams(), + ) + .await + { + Ok(vmms) => vmms, + Err(err) => { + result.errors.push(err); + break; + } + }; + paginator = p.found_batch(&vmms, &|vmm| { + (vmm.sled_id().into_untyped_uuid(), vmm.id) + }); + + self.stop_batch( + &vmms, + &reqwest_client, + &task_opctx, + &mut tasks, + &mut result, + ) + .await; + } + + while let Some(vmm_result) = tasks.join_next().await { + result.merge(vmm_result); + } + + result + } + + /// Stops a batch of VMMs + async fn stop_batch( + &self, + vmms: &[Vmm], + reqwest_client: &ReqwestClient, + opctx: &Arc, + tasks: &mut ParallelTaskSet, + result: &mut VmmStopForUpdateResult, + ) { + for vmms_per_sled in vmms.chunk_by(|a, b| a.sled_id() == b.sled_id()) { + // We only retrieve the first VMM in order to get the sled id. + // The slice is constructed by VMMs with the same sled id, so it's + // fine to just check the first one + let Some(first_vmm) = vmms_per_sled.first() else { + // `chunk_by` never returns an empty slice, so this should never + // be `None`. We handle the case regardless + continue; + }; + let sled_id = first_vmm.sled_id(); + + let vmm_ids: Vec = vmms_per_sled + .iter() + .map(|vmm| PropolisUuid::from_untyped_uuid(vmm.id)) + .collect(); + + slog::info!( + opctx.log, + "Stopping VMMs for update"; + "sled_id" => %sled_id, + "vmms" => ?vmm_ids, + ); + + // Since we're chunking by sled id, each will only appear once per + // batch of VMMs. Because of this, it's ok to start a new sled agent + // client per sled id. + let sled_client = match nexus_networking::sled_client_ext( + &self.datastore, + &opctx, + sled_id, + &opctx.log, + reqwest_client.clone(), + ) + .await + { + Ok(client) => client, + Err(e) => { + slog::error!( + opctx.log, + "Failed to create the sled agent client"; + "sled_id" => %sled_id, + InlineErrorChain::new(&e), + ); + + result.errors.push(e.clone()); + + continue; + } + }; + + for vmm in vmms_per_sled { + let stop = + self.stop_vmm(opctx, sled_client.clone(), vmm.clone()); + if let Some(vmm_result) = tasks.spawn(stop).await { + result.merge(vmm_result); + } + } + } + } + + /// Stops a single VMM. + fn stop_vmm( + &self, + opctx: &Arc, + sled_client: sled_agent_client::Client, + vmm: Vmm, + ) -> impl Future + Send + 'static + use<> + { + let datastore = self.datastore.clone(); + let sagas = self.sagas.clone(); + let opctx = Arc::clone(opctx); + let vmm_id = PropolisUuid::from_untyped_uuid(vmm.id); + let instance_id = InstanceUuid::from_untyped_uuid(vmm.instance_id); + let sled_id = vmm.sled_id(); + + async move { + let mut result = VmmStopForUpdateResult::new(); + let response = match sled_client + .vmm_put_state( + &vmm_id, + &VmmPutStateBody { state: VmmStateRequested::Stopped }, + ) + .await + { + Ok(res) => res.into_inner().updated_runtime.map(Into::into), + Err(e) => { + let err = SledAgentInstanceError(e); + + // First we check if the VMM is there, if it isn't, we + // want to report it as such so we don't have to wait on the + // instance watcher to find and restart this instance. + if err.vmm_gone() { + slog::info!( + opctx.log, + "Failed to mark VMM as Stopped; VMM was already \ + gone. Marking as Failed"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "instance_id" => %instance_id, + InlineErrorChain::new(&err), + ); + + result.errors.push(err.into()); + Some(SledVmmState { + vmm_state: vmm.runtime().transition( + VmmState::Failed( + VmmFailureReason::NoSuchInstance, + ), + ), + migration_in: None, + migration_out: None, + }) + } else { + slog::info!( + opctx.log, + "Failed to mark VMM as Stopped"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "instance_id" => %instance_id, + InlineErrorChain::new(&err), + ); + + result.errors.push(err.into()); + result + .vmms_failed_to_stop_by_sled + .entry(sled_id) + .or_insert_with(|| VmmsBySled { + sled_id, + vmm_ids: BTreeSet::new(), + }) + .vmm_ids + .insert(vmm_id); + + return result; + } + } + }; + + match response { + None => { + slog::debug!( + opctx.log, + "VMM already stopped, no changes to state"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "instance_id" => %instance_id, + ); + } + Some(sled_vmm_state) => { + // Updating the VMM as we have done above is, in practice, + // enough to stop the VMM itself. The instance on the other + // hand, is not updated instantaneously to report the new + // state the VMM is in. + // + // There are a few ways in which the instance will know the + // state of the VMM has changed, and therefore its own state + // should change to reflect the VMM's new state: + // + // - The instance watcher background task continuously + // makes sure the instances' states are correct in + // relation to their VMMs. + // - Sled agent will attempt to publish the new state to + // Nexus when the VMM process exits. + // + // Both of these methods are not immediate, and it will take + // some time for the instance to report itself as `Stopped`. + // To reduce the latency from the time we change the `Vmm`'s + // state to the time this is reflected in the instance, we + // do two things: + // + // - Attempt to write the new VMM state to the db with + // `instance::process_vmm_update`. + // - Execute the instance update saga that + // `instance::process_vmm_update` returns to reconcile + // the instance to the new VMM state. + // + // We don't care if either of the previous two steps fails. + // We are running these steps as an _optimisation_ only. The + // instance state will be updated eventually by either the + // instance watcher task, or sled agent. + try_instance_state_update( + &mut result, + datastore, + sagas, + &opctx, + AssociatedIds { vmm_id, instance_id, sled_id }, + sled_vmm_state.clone(), + ) + .await; + + let SledVmmState { + vmm_state, + migration_in: _, + migration_out: _, + } = sled_vmm_state; + + let VmmRuntimeState { state, generation, time_updated } = + vmm_state; + + match state { + // If we succeeded, one of three things happened: + // + // - The VMM has not stopped yet, but propolis has + // accepted the `put_state` request and reports + // the VMM's state as `Stopping`. + // - A propolis zone does not exist; the VMM is + // reported as `Destroyed`. + // - A VMM should technically never be reported as + // `Stopped` since + // propolis_client::types::InstanceState::Stopped + // is always reported as `VmmState::Stopping`. + // But, if for some reason we get this state, + // it's what we want, so we consider it a + // success. + VmmState::Stopped + | VmmState::Stopping + | VmmState::Destroyed => { + result + .vmms_stopped_by_sled + .entry(sled_id) + .or_insert_with(|| VmmsBySled { + sled_id, + vmm_ids: BTreeSet::new(), + }) + .vmm_ids + .insert(vmm_id); + + slog::debug!( + opctx.log, + "Stopped VMM for sled evacuation during \ + an update"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "vmm_generation" => ?generation, + "instance_id" => %instance_id, + "time_updated" => ?time_updated, + ); + } + // `vmm_put_state` only reports the `VmmState::Failed` + // state when Propolis is alive but does not find a VMM. + // This implies that it restarted and lost the + // previously created VMM. + // + // Additionally, when `vmm_put_state` returns a + // `NO_SUCH_INSTANCE` error, we transition the runtime + // state to `VmmState::Failed` above. + // + // Because of this, it's not necessary to stop it + // anymore, so we don't add it to the + // `vmms_failed_to_stop_by_sled` list. The instance + // reincarnation task will automatically attempt to + // restart any instance that is in a failed state. + VmmState::Failed(reason) => { + let description = reason.description(); + slog::info!( + opctx.log, + "Unable to find VMM. No need to stop it"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "instance_id" => %instance_id, + "reason" => description, + "state" => ?state, + "vmm_generation" => ?generation, + "time_updated" => ?time_updated, + ); + } + VmmState::Migrating + | VmmState::Rebooting + | VmmState::Starting + | VmmState::Running + | VmmState::Creating + | VmmState::SagaUnwound => { + slog::warn!( + opctx.log, + "Failed to mark VMM as Stopped, \ + unexpected state reported"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "state" => ?state, + "vmm_generation" => ?generation, + "instance_id" => %instance_id, + "time_updated" => ?time_updated, + ); + result.errors.push(Error::InternalError { + internal_message: format!( + "failed to mark VMM {} on sled {} as \ + Stopped; unexpected state reported {}", + vmm_id, sled_id, state + ), + }); + } + } + } + } + + result + } + } + + pub(crate) async fn actually_activate( + &mut self, + opctx: &OpContext, + ) -> VmmStopForUpdateStatus { + let VmmStopForUpdateResult { + vmms_stopped_by_sled, + vmms_failed_to_stop_by_sled, + errors, + } = self.stop_all(opctx).await; + + let error_messages: Vec = errors + .iter() + .map(|err| InlineErrorChain::new(err).to_string()) + .collect(); + + if vmms_stopped_by_sled.is_empty() { + slog::info!( + &opctx.log, + "No VMMs were stopped for a sled update"; + ); + } else { + for sled in &vmms_stopped_by_sled { + let VmmsBySled { sled_id, vmm_ids } = sled; + slog::info!( + &opctx.log, + "Stopped {} VMMs marked to stop for a sled update", + vmm_ids.len(); + "sled_id" => %sled_id, + "vmms" => ?vmm_ids, + ); + } + } + + if !vmms_failed_to_stop_by_sled.is_empty() { + for sled in &vmms_failed_to_stop_by_sled { + let VmmsBySled { sled_id, vmm_ids } = sled; + slog::info!( + &opctx.log, + "Failed to stop {} VMMs marked to stop for a sled update", + vmm_ids.len(); + "sled_id" => %sled_id, + "vmms" => ?vmm_ids, + ); + } + } + + if !error_messages.is_empty() { + slog::info!( + &opctx.log, + "Attempting to stop VMMs marked to stop for a sled update \ + returned errors"; + "errors" => ?error_messages, + ); + } + + VmmStopForUpdateStatus { + vmms_stopped_by_sled, + vmms_failed_to_stop_by_sled, + error_messages, + } + } +} + +impl BackgroundTask for VmmStopForUpdate { + fn activate<'a>( + &'a mut self, + opctx: &'a OpContext, + ) -> BoxFuture<'a, serde_json::Value> { + Box::pin(async { + let status = self.actually_activate(opctx).await; + match serde_json::to_value(status) { + Ok(val) => val, + Err(err) => { + json!({ + "error": format!("failed to serialize status: {err}") + }) + } + } + }) + } +} + +// This struct exists solely to avoid a clippy "too_many_arguments" error +struct AssociatedIds { + vmm_id: PropolisUuid, + instance_id: InstanceUuid, + sled_id: SledUuid, +} + +// Attempt to update the instance state. +// +// This function is intended to be run by a caller that does not care whether +// the instance state is actually successfully updated or not. It does not +// Errors are not returned, but rather logged and saved in `VmmStopForUpdateResult`. +async fn try_instance_state_update( + result: &mut VmmStopForUpdateResult, + datastore: Arc, + sagas: Arc, + opctx: &Arc, + ids: AssociatedIds, + sled_vmm_state: SledVmmState, +) { + let AssociatedIds { vmm_id, instance_id, sled_id } = ids; + + // Attempt to write the new VMM state to the db. + let saga = match instance::process_vmm_update( + &datastore, + opctx, + vmm_id, + &sled_vmm_state, + ) + .await + { + // As mentioned earlier, it's not a huge deal if we fail here. The + // instance state will be updated eventually by the instance watcher + // task or sled agent. + Err(err) => { + slog::warn!( + opctx.log, + "Failed to record new VMM state"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "state" => ?sled_vmm_state, + "instance_id" => %instance_id, + InlineErrorChain::new(&err), + ); + result.errors.push(err); + return; + } + // The new state would like an instance-update saga, so we'll try to + // run it. + Ok(Some((_, saga))) => saga, + // The new state does not require an update saga, nothing left to do + // here. + Ok(None) => return, + }; + let (saga_id, completed) = match sagas.saga_run(saga).await { + Ok(s) => s, + Err(err) => { + slog::warn!( + opctx.log, + "Instance update saga could not be started"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "instance_id" => %instance_id, + InlineErrorChain::new(&err), + ); + result.errors.push(err); + return; + } + }; + if let Err(err) = completed.await { + slog::warn!( + opctx.log, + "Instance update saga failed"; + "sled_id" => %sled_id, + "vmm_id" => %vmm_id, + "saga_id" => %saga_id, + "instance_id" => %instance_id, + InlineErrorChain::new(&err), + ); + result.errors.push(err); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::app::sagas::test_helpers::instance_fetch; + use crate::app::sagas::test_helpers::instance_simulate; + use crate::app::sagas::test_helpers::instance_wait_for_state; + use crate::app::sagas::test_helpers::test_opctx; + use crate::app::sagas::test_helpers::wait_for_running_sagas; + use async_bb8_diesel::AsyncRunQueryDsl; + use chrono::Utc; + use diesel::ExpressionMethods; + use diesel::QueryDsl; + use diesel::SelectableHelper; + use nexus_db_lookup::AsyncConnection; + use nexus_db_model::Generation; + use nexus_db_model::InstanceState; + use nexus_db_model::Vmm; + use nexus_db_model::VmmCpuPlatform; + use nexus_db_model::VmmFailureReason; + use nexus_db_model::VmmState as DbVmmState; + use nexus_db_model::to_db_typed_generation; + use nexus_test_utils::ControlPlaneBuilder; + use nexus_test_utils::resource_helpers::create_anti_affinity_group; + use nexus_test_utils::resource_helpers::create_default_ip_pools; + use nexus_test_utils::resource_helpers::create_project; + use nexus_test_utils::resource_helpers::object_create; + use nexus_types::external_api::instance::InstanceCreate; + use nexus_types_versions::latest::instance::Instance; + use nexus_types_versions::latest::instance::InstanceNetworkInterfaceAttachment; + use nexus_types_versions::latest::instance::InstanceSelector; + use omicron_common::api::external::ByteCount; + use omicron_common::api::external::IdentityMetadataCreateParams; + use omicron_common::api::external::NameOrId; + use omicron_generation_kinds::UpdateDispositionGeneration; + use omicron_uuid_kinds::InstanceUuid; + use omicron_uuid_kinds::SledUuid; + use std::collections::BTreeMap; + use std::collections::BTreeSet; + use std::time::Duration; + use uuid::Uuid; + + type ControlPlaneTestContext = + nexus_test_utils::ControlPlaneTestContext; + + const PROJECT_NAME: &str = "vmm-stop-for-update"; + + // In order to stop a VMM, it needs to be registered with the simulated + // sled agent. We keep a record of the states of the real VMMs we will be + // stopping. + // + // The label is used to name instances and groups. + const REAL_VMM_STATES: [(&str, DbVmmState); 3] = [ + ("starting", DbVmmState::Starting), + ("running", DbVmmState::Running), + ("rebooting", DbVmmState::Rebooting), + ]; + + async fn create_started_instance( + cptestctx: &ControlPlaneTestContext, + name: &str, + anti_affinity_group: &str, + ) -> InstanceUuid { + let created = object_create::<_, Instance>( + &cptestctx.external_client, + &format!("/v1/instances?project={PROJECT_NAME}"), + &InstanceCreate { + identity: IdentityMetadataCreateParams { + name: name.parse().unwrap(), + description: format!("instance {name}"), + }, + ncpus: 1i64.try_into().unwrap(), + memory: ByteCount::from_gibibytes_u32(1), + hostname: "myhostname".try_into().unwrap(), + user_data: Vec::new(), + network_interfaces: InstanceNetworkInterfaceAttachment::None, + external_ips: Vec::new(), + disks: Vec::new(), + boot_disk: None, + cpu_platform: None, + ssh_public_keys: None, + start: true, + auto_restart_policy: None, + anti_affinity_groups: vec![NameOrId::Name( + anti_affinity_group.parse().unwrap(), + )], + multicast_groups: Vec::new(), + enable_jumbo_frames: false, + }, + ) + .await; + InstanceUuid::from_untyped_uuid(created.identity.id) + } + + async fn reboot_instance( + cptestctx: &ControlPlaneTestContext, + instance_id: InstanceUuid, + ) { + let nexus = &cptestctx.server.server_context().nexus; + let opctx = test_opctx(cptestctx); + let lookup = nexus + .instance_lookup( + &opctx, + InstanceSelector { + project: None, + instance: NameOrId::from(instance_id.into_untyped_uuid()), + }, + ) + .expect("instance lookup should succeed"); + nexus + .instance_reboot(&opctx, &lookup) + .await + .expect("instance should reboot"); + } + + async fn active_vmm( + cptestctx: &ControlPlaneTestContext, + instance_id: InstanceUuid, + ) -> Vmm { + instance_fetch(cptestctx, instance_id) + .await + .vmm() + .clone() + .expect("instance should have an active VMM") + } + + async fn mark_vmm_stopped_for_update( + conn: &AsyncConnection, + vmm_id: Uuid, + generation: UpdateDispositionGeneration, + ) { + use nexus_db_schema::schema::vmm::dsl; + + let updated = diesel::update(dsl::vmm) + .filter(dsl::id.eq(vmm_id)) + .set( + dsl::stop_for_update_disposition_generation + .eq(Some(to_db_typed_generation(generation))), + ) + .execute_async(conn) + .await + .expect("VMM should be marked"); + assert_eq!(updated, 1, "exactly one VMM should be marked"); + } + + // Inserts a VMM row without sled agent knowing about it. This is used for + // VMMs in states that we don't expect to stop. + async fn insert_vmm_in_state( + datastore: &DataStore, + opctx: &OpContext, + sled_id: SledUuid, + state: DbVmmState, + marker: Option, + ) -> Vmm { + let failure_reason = (state == DbVmmState::Failed) + .then_some(VmmFailureReason::FromSledAgent); + datastore + .vmm_insert( + opctx, + Vmm { + id: Uuid::new_v4(), + time_created: Utc::now(), + time_deleted: None, + instance_id: Uuid::new_v4(), + sled_id: sled_id.into(), + propolis_ip: "10.1.9.32".parse().unwrap(), + propolis_port: 420.into(), + cpu_platform: VmmCpuPlatform::SledDefault, + time_state_updated: Utc::now(), + generation: Generation::new(), + state, + failure_reason, + stop_for_update_disposition_generation: marker + .map(Into::into), + }, + ) + .await + .expect("VMM should be inserted") + } + + // `vmm_fetch` does not retrieve rows with a non-null `time_deleted`, so we + // need a little helper that does. + async fn fetch_vmms( + conn: &AsyncConnection, + ids: &[Uuid], + ) -> BTreeMap { + use nexus_db_schema::schema::vmm::dsl; + + dsl::vmm + .filter(dsl::id.eq_any(ids.to_vec())) + .select(Vmm::as_select()) + .load_async::(conn) + .await + .expect("VMMs should be fetched") + .into_iter() + .map(|vmm| (vmm.id, vmm)) + .collect() + } + + fn vmm_ids_in_all_sleds( + vmms_by_sled: &IdOrdMap, + ) -> BTreeSet { + vmms_by_sled + .iter() + .flat_map(|sled| { + sled.vmm_ids.iter().map(|id| id.into_untyped_uuid()) + }) + .collect() + } + + #[tokio::test] + async fn test_vmm_stop_for_update_activation() { + // Set up the test environment. + // + // We want two sleds to test that the task stops VMMs across multiple + // sleds. Additionally, to avoid flakiness, we increase the intervals + // for background tasks that could potentially modify VMM rows while + // this test is running. + let cptestctx = + ControlPlaneBuilder::new("test_vmm_stop_for_update_activation") + .with_extra_sled_agents(1) + .customize_nexus_config(&|config| { + let one_day = Duration::from_secs(86_400); + let tasks = &mut config.pkg.background_tasks; + tasks.instance_watcher.period_secs = one_day; + tasks.abandoned_vmm_reaper.period_secs = one_day; + tasks.vmm_mark_stop_for_update.period_secs = one_day; + tasks.vmm_stop_for_update.period_secs = one_day; + }) + .start::() + .await; + + let client = &cptestctx.external_client; + let datastore = + cptestctx.server.server_context().nexus.datastore().clone(); + let nexus = cptestctx.server.server_context().nexus.clone(); + let opctx = test_opctx(&cptestctx); + let sled_a = cptestctx.first_sled_id(); + let sled_b = cptestctx.second_sled_id(); + let generation = UpdateDispositionGeneration::from(1); + + create_default_ip_pools(client).await; + create_project(client, PROJECT_NAME).await; + + // The simulated agent can only stop VMMs that are registered with it. + // So, for each VMM that we intend to stop, we create a real instance + // via the external API. + // + // For each state in `REAL_VMM_STATES`, two instances are created per + // anti-affinity group. This ensures that both sleds have the same + // amount of instances in each state. + let conn = datastore.pool_connection_for_tests().await.unwrap(); + let mut instance_ids = Vec::new(); + let mut vmm_ids = Vec::new(); + for (label, state) in REAL_VMM_STATES { + let group = format!("group-{label}"); + create_anti_affinity_group(client, PROJECT_NAME, &group).await; + for i in 0..2 { + let id = create_started_instance( + &cptestctx, + &format!("{label}-{i}"), + &group, + ) + .await; + match state { + // A new VMM in a simulated environment stays in `Starting` + // until the "poke" API is called, so we can confidently + // test a VMM in this state. + DbVmmState::Starting => {} + DbVmmState::Running => { + instance_simulate(&cptestctx, &id).await; + } + DbVmmState::Rebooting => { + instance_simulate(&cptestctx, &id).await; + // Like above, we can keep a VMM in `Rebooting` if we + // don't call the "poke" API + reboot_instance(&cptestctx, id).await; + } + _ => unreachable!("{state:?} is not in REAL_VMM_STATES"), + } + let vmm = active_vmm(&cptestctx, id).await; + assert_eq!(vmm.state, state, "VMM {} is not {state:?}", vmm.id); + // Mark them to be stopped by the task + mark_vmm_stopped_for_update(&conn, vmm.id, generation).await; + instance_ids.push(id); + vmm_ids.push(vmm.id); + } + } + + // For all other states that shouldn't be stopped by the task, we insert + // rows directly in the DB. The task should skip them entirely, so they + // don't need to be registered with sled agent. + // + // We insert marked non-stoppable VMMs. This ensures that even if the + // task to mark VMMs marks a VMM that is in a state that it shouldn't be + // stopped in, this task skips it entirely. + for sled_id in [sled_a, sled_b] { + for &state in DbVmmState::ALL_STATES { + let unmarked = insert_vmm_in_state( + &datastore, &opctx, sled_id, state, None, + ) + .await; + vmm_ids.push(unmarked.id); + + if !REAL_VMM_STATES.iter().any(|(_, real)| *real == state) { + let marked = insert_vmm_in_state( + &datastore, + &opctx, + sled_id, + state, + Some(generation), + ) + .await; + vmm_ids.push(marked.id); + } + } + } + + let vmms = fetch_vmms(&conn, &vmm_ids).await; + + // There should be 40 VMM rows: + // - 6 VMMs registered with sled agent (one in each of the 3 + // `REAL_VMM_STATES` on each of the 2 sleds) + // - 20 unmarked VMMs (one in every state on each sled) + // - 14 marked VMMs (one in each of the states *not* in + // `REAL_VMM_STATES` on each sled) + assert_eq!(vmms.len(), 40); + + // Collect which VMMs we expect the task will stop and which it won't. + let should_be_stopped = |vmm: &Vmm| { + vmm.stop_for_update_disposition_generation.is_some() + && DbVmmState::SHOULD_STOP_FOR_EVACUATION.contains(&vmm.state) + }; + let (expected_rows_to_be_stopped, expected_rows_to_not_be_stopped): ( + Vec, + Vec, + ) = vmms.values().cloned().partition(should_be_stopped); + + assert_eq!(expected_rows_to_be_stopped.len(), 6); + + // Confirm there is one of each stoppable state per sled in + // `expected_rows_to_be_stopped` + for sled_id in [sled_a, sled_b] { + for (_, state) in REAL_VMM_STATES { + let count = expected_rows_to_be_stopped + .iter() + .filter(|vmm| { + vmm.sled_id() == sled_id && vmm.state == state + }) + .count(); + assert_eq!( + count, 1, + "sled {sled_id} should have one marked VMM in {state:?}" + ); + } + } + + // Run the task. + let mut task = + VmmStopForUpdate::new(datastore.clone(), nexus.sagas.clone()); + let status = task.actually_activate(&opctx).await; + + // No errors are expected + assert_eq!( + status.error_messages.len(), + 0, + "unexpected errors: {:?}", + status.error_messages + ); + + // We stopped the VMMs we expected to stop + let stopped_count: usize = status + .vmms_stopped_by_sled + .iter() + .map(|sled| sled.vmm_ids.len()) + .sum(); + assert_eq!(stopped_count, expected_rows_to_be_stopped.len()); + + let expected_stopped_ids: BTreeSet = + expected_rows_to_be_stopped.iter().map(|vmm| vmm.id).collect(); + assert_eq!( + vmm_ids_in_all_sleds(&status.vmms_stopped_by_sled), + expected_stopped_ids + ); + + // No VMMs are expected to fail to stop + let failed_to_stop_count: usize = status + .vmms_failed_to_stop_by_sled + .iter() + .map(|sled| sled.vmm_ids.len()) + .sum(); + assert_eq!(failed_to_stop_count, 0); + + // No errors are expected + assert_eq!( + status.error_messages.len(), + 0, + "unexpected errors: {:?}", + status.error_messages + ); + + // Poke each simulated instance so each sled agent reports back to Nexus + // and wait until each one reports `NoVmm` + for id in &instance_ids { + instance_simulate(&cptestctx, id).await; + instance_wait_for_state(&cptestctx, *id, InstanceState::NoVmm) + .await; + } + // Before fetching the VMMs again, make sure all instance update sagas + // are finished + wait_for_running_sagas(&cptestctx).await; + let vmms_after = fetch_vmms(&conn, &vmm_ids).await; + assert_eq!(vmms_after.len(), 40); + + // Every VMM we expected to stop should be destroyed at this point + for expected in &expected_rows_to_be_stopped { + let actual = &vmms_after[&expected.id]; + assert_eq!( + actual.state, + DbVmmState::Destroyed, + "VMM {} on sled {} was not stopped", + expected.id, + expected.sled_id(), + ); + assert!( + actual.time_deleted.is_some(), + "VMM {} was not deleted", + expected.id, + ); + assert_eq!( + actual.stop_for_update_disposition_generation, + expected.stop_for_update_disposition_generation, + ); + } + + // Make sure all other rows are unchanged + for expected in &expected_rows_to_not_be_stopped { + assert_eq!( + &vmms_after[&expected.id], + expected, + "VMM {} on sled {} in state {:?} changed unexpectedly", + expected.id, + expected.sled_id(), + expected.state, + ); + } + + // Run the task again. + let status_second_run = task.actually_activate(&opctx).await; + + // No new VMMs were marked to be stopped so nothing is stopped + assert!(status_second_run.vmms_stopped_by_sled.is_empty()); + + // Nothing changed, so we expect no errors or failures + assert!(status_second_run.vmms_failed_to_stop_by_sled.is_empty()); + assert!(status_second_run.error_messages.is_empty()); + + // The second run makes no changes to the VMMs + let vmms_after_second_run = fetch_vmms(&conn, &vmm_ids).await; + assert_eq!(vmms_after_second_run, vmms_after); + + cptestctx.teardown().await; + } +} diff --git a/nexus/src/app/sagas/instance_update/mod.rs b/nexus/src/app/sagas/instance_update/mod.rs index 4d99e6592a4..444b5a818dc 100644 --- a/nexus/src/app/sagas/instance_update/mod.rs +++ b/nexus/src/app/sagas/instance_update/mod.rs @@ -555,8 +555,8 @@ impl UpdatesRequired { let id = PropolisUuid::from_untyped_uuid(active_vmm.id); // Unlink the active VMM ID. If the active VMM was destroyed // because a migration out completed, the next block, which - // handles migration updates, will set this to the new VMM's ID, - // instead. + // handles migration updates, will set this to the new VMM's + // ID, instead. new_runtime.propolis_id = None; update_required = true; @@ -567,10 +567,14 @@ impl UpdatesRequired { } else if snapshot.instance.intended_state == InstanceIntendedState::Running && snapshot.migration.is_none() + && active_vmm + .stop_for_update_disposition_generation + .is_none() { // Did the active VMM shut itself down, when it was // intended to be running *and* we were not migrating - // out? If so, update the instance's intended state to + // out, or destroying it intentionally for an update? + // If so, update the instance's intended state to // indicate that it stopped itself. new_intent = Some(InstanceIntendedState::GuestShutdown); info!( diff --git a/nexus/tests/config.test.toml b/nexus/tests/config.test.toml index abea37101e2..aca224368ea 100644 --- a/nexus/tests/config.test.toml +++ b/nexus/tests/config.test.toml @@ -151,6 +151,7 @@ instance_watcher.period_secs = 30 service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 +vmm_stop_for_update.period_secs = 300 saga_recovery.period_secs = 600 lookup_region_port.period_secs = 60 # The purpose of the `instance-updater` background task is to ensure that update diff --git a/nexus/types/src/internal_api/background.rs b/nexus/types/src/internal_api/background.rs index d3e7a1f24ce..b6f63e6fc26 100644 --- a/nexus/types/src/internal_api/background.rs +++ b/nexus/types/src/internal_api/background.rs @@ -15,6 +15,7 @@ use omicron_uuid_kinds::AlertReceiverUuid; use omicron_uuid_kinds::AlertUuid; use omicron_uuid_kinds::BlueprintUuid; use omicron_uuid_kinds::CollectionUuid; +use omicron_uuid_kinds::PropolisUuid; use omicron_uuid_kinds::RackUuid; use omicron_uuid_kinds::SitrepUuid; use omicron_uuid_kinds::SledUuid; @@ -27,6 +28,7 @@ use serde::Serialize; use sled_agent_types::early_networking::SwitchSlot; use slog::Key; use std::collections::BTreeMap; +use std::collections::BTreeSet; use std::collections::HashMap; use std::collections::VecDeque; use std::net::Ipv6Addr; @@ -106,6 +108,36 @@ pub struct AbandonedVmmReaperStatus { pub errors: Vec, } +/// VMMs on a single sled +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct VmmsBySled { + pub sled_id: SledUuid, + pub vmm_ids: BTreeSet, +} + +impl IdOrdItem for VmmsBySled { + type Key<'a> = SledUuid; + + fn key(&self) -> Self::Key<'_> { + self.sled_id + } + + id_upcast!(); +} + +/// The status of a `vmm_stop_for_update` background task activation. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct VmmStopForUpdateStatus { + /// Number of VMMs stopped in this activation, keyed by the sled they were + /// running on. + pub vmms_stopped_by_sled: IdOrdMap, + /// Number of VMMs that failed to stop in this activation, keyed by the sled + /// they are running on. + pub vmms_failed_to_stop_by_sled: IdOrdMap, + /// Errors encountered during this activation, if any. + pub error_messages: Vec, +} + /// The status of an `instance_updater` background task activation. #[derive(Serialize, Deserialize, Default, Debug, PartialEq, Eq)] pub struct InstanceUpdaterStatus { @@ -229,6 +261,10 @@ pub enum ReincarnationReason { Failed, /// A previous instance-start saga for this instance has failed. SagaUnwound, + // TODO-K: In a follow up PR add a StoppedForUpdate reason, so the instances + // we stopped during sled evacuation actually get restarted + // + // Tracking issue: https://github.com/oxidecomputer/omicron/issues/11169 } impl std::fmt::Display for ReincarnationReason { diff --git a/smf/nexus/multi-sled/config-partial.toml b/smf/nexus/multi-sled/config-partial.toml index ebc221cf69a..5c8b58bebbc 100644 --- a/smf/nexus/multi-sled/config-partial.toml +++ b/smf/nexus/multi-sled/config-partial.toml @@ -67,6 +67,7 @@ service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 instance_watcher.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 +vmm_stop_for_update.period_secs = 60 saga_recovery.period_secs = 600 lookup_region_port.period_secs = 60 instance_updater.period_secs = 30 diff --git a/smf/nexus/single-sled/config-partial.toml b/smf/nexus/single-sled/config-partial.toml index 5769ae89f23..34b0badd2d8 100644 --- a/smf/nexus/single-sled/config-partial.toml +++ b/smf/nexus/single-sled/config-partial.toml @@ -67,6 +67,7 @@ service_firewall_propagation.period_secs = 300 v2p_mapping_propagation.period_secs = 30 instance_watcher.period_secs = 30 abandoned_vmm_reaper.period_secs = 60 +vmm_stop_for_update.period_secs = 60 saga_recovery.period_secs = 600 lookup_region_port.period_secs = 60 instance_updater.period_secs = 30