[reconfigurator] Executor internal cleanup around iterators (#7722)

Builds on #7713, and is followup from
#7713 (comment).
In #7652 I changed all the executor substeps to take iterators instead
of `&BTreeMap` references that no longer existed, but that introduced a
weird split where the top-level caller had to filter the blueprint down
to just the items that the inner functions expected. @smklein pointed
out one place where the inner code was being extra defensive, which was
just more confusing than anything.

This PR removes that split: the top-level executor now always passes a
full `&Blueprint` down, and the inner modules are responsible for doing
their own filtering as appropriate. To easy testing, I kept the versions
that take an iterator of already-filtered items as private `*_impl`
functions that the new functions-that-take-a-full-`Blueprint` themselves
call too.

Author: jgallagher

nexus/reconfigurator/execution/src/clickhouse.rs

@@ -22,7 +22,9 @@ use futures::future::Either;
 use futures::stream::FuturesUnordered;
 use futures::stream::StreamExt;
 use nexus_db_queries::context::OpContext;
+use nexus_types::deployment::Blueprint;
 use nexus_types::deployment::BlueprintZoneConfig;
+use nexus_types::deployment::BlueprintZoneDisposition;
 use nexus_types::deployment::ClickhouseClusterConfig;
 use omicron_common::address::CLICKHOUSE_ADMIN_PORT;
 use omicron_uuid_kinds::OmicronZoneUuid;
@@ -38,7 +40,20 @@ use std::str::FromStr;
 
 const CLICKHOUSE_DATA_DIR: &str = "/data";
 
-pub(crate) async fn deploy_nodes<'a, I>(
+pub(crate) async fn deploy_nodes(
+    opctx: &OpContext,
+    blueprint: &Blueprint,
+    clickhouse_cluster_config: &ClickhouseClusterConfig,
+) -> Result<(), Vec<anyhow::Error>> {
+    deploy_nodes_impl(
+        opctx,
+        blueprint.all_omicron_zones(BlueprintZoneDisposition::any),
+        clickhouse_cluster_config,
+    )
+    .await
+}
+
+async fn deploy_nodes_impl<'a, I>(
     opctx: &OpContext,
     zones: I,
     clickhouse_cluster_config: &ClickhouseClusterConfig,
@@ -186,7 +201,20 @@ where
     Ok(())
 }
 
-pub(crate) async fn deploy_single_node<'a, I>(
+pub(crate) async fn deploy_single_node(
+    opctx: &OpContext,
+    blueprint: &Blueprint,
+) -> Result<(), anyhow::Error> {
+    deploy_single_node_impl(
+        opctx,
+        blueprint
+            .all_omicron_zones(BlueprintZoneDisposition::is_in_service)
+            .filter(|(_, z)| z.zone_type.is_clickhouse()),
+    )
+    .await
+}
+
+async fn deploy_single_node_impl<'a, I>(
     opctx: &OpContext,
     mut zones: I,
 ) -> Result<(), anyhow::Error>
@@ -330,7 +358,6 @@ mod test {
     use clickhouse_admin_types::ServerId;
     use nexus_sled_agent_shared::inventory::OmicronZoneDataset;
     use nexus_types::deployment::BlueprintZoneConfig;
-    use nexus_types::deployment::BlueprintZoneDisposition;
     use nexus_types::deployment::BlueprintZoneImageSource;
     use nexus_types::deployment::BlueprintZoneType;
     use nexus_types::deployment::blueprint_zone_type;

nexus/reconfigurator/execution/src/datasets.rs

@@ -11,6 +11,7 @@ use anyhow::anyhow;
 use futures::StreamExt;
 use futures::stream;
 use nexus_db_queries::context::OpContext;
+use nexus_types::deployment::Blueprint;
 use nexus_types::deployment::BlueprintDatasetsConfig;
 use omicron_common::disk::DatasetsConfig;
 use omicron_uuid_kinds::GenericUuid;
@@ -22,7 +23,23 @@ use std::collections::BTreeMap;
 
 /// Idempotently ensures that the specified datasets are deployed to the
 /// corresponding sleds
-pub(crate) async fn deploy_datasets<'a, I>(
+pub(crate) async fn deploy_datasets(
+    opctx: &OpContext,
+    sleds_by_id: &BTreeMap<SledUuid, Sled>,
+    blueprint: &Blueprint,
+) -> Result<(), Vec<anyhow::Error>> {
+    deploy_datasets_impl(
+        opctx,
+        sleds_by_id,
+        blueprint
+            .sleds
+            .iter()
+            .map(|(sled_id, sled)| (*sled_id, &sled.datasets_config)),
+    )
+    .await
+}
+
+async fn deploy_datasets_impl<'a, I>(
     opctx: &OpContext,
     sleds_by_id: &BTreeMap<SledUuid, Sled>,
     sled_configs: I,
@@ -184,7 +201,7 @@ mod tests {
         let sled_configs = [(sim_sled_agent.id, &datasets_config)];
 
         // Give the simulated sled agent a configuration to deploy
-        deploy_datasets(&opctx, &sleds_by_id, sled_configs.into_iter())
+        deploy_datasets_impl(&opctx, &sleds_by_id, sled_configs.into_iter())
             .await
             .expect("Deploying datasets should have succeeded");
 

nexus/reconfigurator/execution/src/lib.rs

@@ -11,11 +11,8 @@ use internal_dns_resolver::Resolver;
 use nexus_db_queries::context::OpContext;
 use nexus_db_queries::db::DataStore;
 use nexus_types::deployment::Blueprint;
-use nexus_types::deployment::BlueprintPhysicalDiskDisposition;
-use nexus_types::deployment::BlueprintZoneDisposition;
 use nexus_types::deployment::SledFilter;
 use nexus_types::deployment::execution::*;
-use nexus_types::external_api::views::SledState;
 use nexus_types::identity::Asset;
 use omicron_uuid_kinds::GenericUuid;
 use omicron_uuid_kinds::OmicronZoneUuid;
@@ -246,7 +243,7 @@ fn register_zone_external_networking_step<'a>(
             move |_cx| async move {
                 datastore
                     .blueprint_ensure_external_networking_resources(
-                        &opctx, blueprint,
+                        opctx, blueprint,
                     )
                     .await
                     .map_err(|err| anyhow!(err))?;
@@ -271,7 +268,7 @@ fn register_support_bundle_failure_step<'a>(
              an expunged disk or sled",
             move |_cx| async move {
                 let res = match datastore
-                    .support_bundle_fail_expunged(&opctx, blueprint, nexus_id)
+                    .support_bundle_fail_expunged(opctx, blueprint, nexus_id)
                     .await
                 {
                     Ok(report) => StepSuccess::new(())
@@ -298,7 +295,7 @@ fn register_sled_list_step<'a>(
             "Fetch sled list",
             move |_cx| async move {
                 let sleds_by_id: BTreeMap<SledUuid, _> = datastore
-                    .sled_list_all_batched(&opctx, SledFilter::InService)
+                    .sled_list_all_batched(opctx, SledFilter::InService)
                     .await
                     .context("listing all sleds")?
                     .into_iter()
@@ -329,7 +326,7 @@ fn register_deploy_disks_step<'a>(
             move |cx| async move {
                 let sleds_by_id = sleds.into_value(cx.token()).await;
                 let res = omicron_physical_disks::deploy_disks(
-                    &opctx,
+                    opctx,
                     &sleds_by_id,
                     blueprint
                         .sleds
@@ -356,15 +353,10 @@ fn register_deploy_datasets_step<'a>(
             "Deploy datasets",
             move |cx| async move {
                 let sleds_by_id = sleds.into_value(cx.token()).await;
-                let res = datasets::deploy_datasets(
-                    &opctx,
-                    &sleds_by_id,
-                    blueprint.sleds.iter().map(|(sled_id, sled)| {
-                        (*sled_id, &sled.datasets_config)
-                    }),
-                )
-                .await
-                .map_err(merge_anyhow_list);
+                let res =
+                    datasets::deploy_datasets(opctx, &sleds_by_id, blueprint)
+                        .await
+                        .map_err(merge_anyhow_list);
                 Ok(map_err_to_step_warning(res))
             },
         )
@@ -384,7 +376,7 @@ fn register_deploy_zones_step<'a>(
             move |cx| async move {
                 let sleds_by_id = sleds.into_value(cx.token()).await;
                 let res = omicron_zones::deploy_zones(
-                    &opctx,
+                    opctx,
                     &sleds_by_id,
                     blueprint
                         .sleds
@@ -418,9 +410,9 @@ fn register_plumb_firewall_rules_step<'a>(
             move |_cx| async move {
                 let res = nexus_networking::plumb_service_firewall_rules(
                     datastore,
-                    &opctx,
+                    opctx,
                     &[],
-                    &opctx,
+                    opctx,
                     &opctx.log,
                 )
                 .await
@@ -448,7 +440,7 @@ fn register_dns_records_step<'a>(
                 let sleds_by_id = sleds.into_value(cx.token()).await;
 
                 let res = dns::deploy_dns(
-                    &opctx,
+                    opctx,
                     datastore,
                     nexus_id.to_string(),
                     blueprint,
@@ -476,12 +468,7 @@ fn register_cleanup_expunged_zones_step<'a>(
             "Cleanup expunged zones",
             move |_cx| async move {
                 let res = omicron_zones::clean_up_expunged_zones(
-                    &opctx,
-                    datastore,
-                    resolver,
-                    blueprint.all_omicron_zones(
-                        BlueprintZoneDisposition::is_ready_for_cleanup,
-                    ),
+                    opctx, datastore, resolver, blueprint,
                 )
                 .await
                 .map_err(merge_anyhow_list);
@@ -502,19 +489,10 @@ fn register_decommission_sleds_step<'a>(
             ExecutionStepId::Cleanup,
             "Decommission sleds",
             move |_cx| async move {
-                let res = sled_state::decommission_sleds(
-                    &opctx,
-                    datastore,
-                    blueprint
-                        .sleds
-                        .iter()
-                        .filter(|(_, sled)| {
-                            sled.state == SledState::Decommissioned
-                        })
-                        .map(|(&sled_id, _)| sled_id),
-                )
-                .await
-                .map_err(merge_anyhow_list);
+                let res =
+                    sled_state::decommission_sleds(opctx, datastore, blueprint)
+                        .await
+                        .map_err(merge_anyhow_list);
                 Ok(map_err_to_step_warning(res))
             },
         )
@@ -533,11 +511,7 @@ fn register_decommission_disks_step<'a>(
             "Decommission expunged disks",
             move |_cx| async move {
                 let res = omicron_physical_disks::decommission_expunged_disks(
-                    &opctx,
-                    datastore,
-                    blueprint
-                        .all_omicron_disks(BlueprintPhysicalDiskDisposition::is_ready_for_cleanup)
-                        .map(|(sled_id, config)| (sled_id, config.id)),
+                    opctx, datastore, blueprint,
                 )
                 .await
                 .map_err(merge_anyhow_list);
@@ -561,9 +535,8 @@ fn register_deploy_clickhouse_cluster_nodes_step<'a>(
                     &blueprint.clickhouse_cluster_config
                 {
                     let res = clickhouse::deploy_nodes(
-                        &opctx,
-                        blueprint
-                            .all_omicron_zones(BlueprintZoneDisposition::any),
+                        opctx,
+                        blueprint,
                         &clickhouse_cluster_config,
                     )
                     .await
@@ -587,15 +560,8 @@ fn register_deploy_clickhouse_single_node_step<'a>(
             ExecutionStepId::Ensure,
             "Deploy single-node clickhouse cluster",
             move |_cx| async move {
-                let res = clickhouse::deploy_single_node(
-                    &opctx,
-                    blueprint
-                        .all_omicron_zones(
-                            BlueprintZoneDisposition::is_in_service,
-                        )
-                        .filter(|(_, z)| z.zone_type.is_clickhouse()),
-                )
-                .await;
+                let res =
+                    clickhouse::deploy_single_node(opctx, blueprint).await;
                 Ok(map_err_to_step_warning(res))
             },
         )
@@ -620,7 +586,7 @@ fn register_reassign_sagas_step<'a>(
                 // 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,
+                    opctx, datastore, blueprint, sec_id,
                 )
                 .await
                 .context("failed to re-assign sagas");
@@ -649,7 +615,7 @@ fn register_cockroachdb_settings_step<'a>(
             "Ensure CockroachDB settings",
             move |_cx| async move {
                 let res =
-                    cockroachdb::ensure_settings(&opctx, datastore, blueprint)
+                    cockroachdb::ensure_settings(opctx, datastore, blueprint)
                         .await;
                 Ok(map_err_to_step_warning(res))
             },

nexus/reconfigurator/execution/src/omicron_physical_disks.rs

@@ -11,6 +11,8 @@ use futures::StreamExt;
 use futures::stream;
 use nexus_db_queries::context::OpContext;
 use nexus_db_queries::db::DataStore;
+use nexus_types::deployment::Blueprint;
+use nexus_types::deployment::BlueprintPhysicalDiskDisposition;
 use nexus_types::deployment::BlueprintPhysicalDisksConfig;
 use omicron_uuid_kinds::GenericUuid;
 use omicron_uuid_kinds::PhysicalDiskUuid;
@@ -104,6 +106,23 @@ where
 
 /// Decommissions all disks which are currently expunged.
 pub(crate) async fn decommission_expunged_disks(
+    opctx: &OpContext,
+    datastore: &DataStore,
+    blueprint: &Blueprint,
+) -> Result<(), Vec<anyhow::Error>> {
+    decommission_expunged_disks_impl(
+        opctx,
+        datastore,
+        blueprint
+            .all_omicron_disks(
+                BlueprintPhysicalDiskDisposition::is_ready_for_cleanup,
+            )
+            .map(|(sled_id, config)| (sled_id, config.id)),
+    )
+    .await
+}
+
+async fn decommission_expunged_disks_impl(
     opctx: &OpContext,
     datastore: &DataStore,
     expunged_disks: impl Iterator<Item = (SledUuid, PhysicalDiskUuid)>,
@@ -642,7 +661,7 @@ mod test {
         assert_eq!(d.disk_state, PhysicalDiskState::Active);
         assert_eq!(d.disk_policy, PhysicalDiskPolicy::InService);
 
-        super::decommission_expunged_disks(
+        super::decommission_expunged_disks_impl(
             &opctx,
             &datastore,
             [(sled_id, disk_to_decommission)].into_iter(),