Make the planner watch for inventory collections

Author: plotnick

dev-tools/omdb/tests/successes.out

@@ -528,7 +528,7 @@ task: "bfd_manager"
 task: "blueprint_planner"
   configured period: every <REDACTED_DURATION>m
   currently executing: no
-  last completed activation: <REDACTED ITERATIONS>, triggered by a periodic timer firing
+  last completed activation: <REDACTED ITERATIONS>, triggered by a dependent task completing
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
     last completion reported error: no target blueprint to use as parent for planning
 
@@ -1054,7 +1054,7 @@ task: "bfd_manager"
 task: "blueprint_planner"
   configured period: every <REDACTED_DURATION>m
   currently executing: no
-  last completed activation: <REDACTED ITERATIONS>, triggered by a periodic timer firing
+  last completed activation: <REDACTED ITERATIONS>, triggered by a dependent task completing
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
     last completion reported error: no target blueprint to use as parent for planning
 

nexus/src/app/background/init.rs

@@ -418,38 +418,12 @@ impl BackgroundTasksInitializer {
         };
 
         // Background task: blueprint loader
+        //
+        // Registration is below so that it can watch the planner.
         let blueprint_loader =
             blueprint_load::TargetBlueprintLoader::new(datastore.clone());
         let rx_blueprint = blueprint_loader.watcher();
 
-        // Background task: blueprint planner
-        let blueprint_planner = blueprint_planner::BlueprintPlanner::new(
-            datastore.clone(),
-            false,
-            rx_blueprint.clone(),
-        );
-        let rx_planner = blueprint_planner.watcher();
-        driver.register(TaskDefinition {
-            name: "blueprint_planner",
-            description: "Updates the target blueprint",
-            period: config.blueprints.period_secs_plan,
-            task_impl: Box::new(blueprint_planner),
-            opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(rx_blueprint.clone())],
-            activator: task_blueprint_planner,
-        });
-
-        // The planner notifies the loader when it has generated a new target blueprint.
-        driver.register(TaskDefinition {
-            name: "blueprint_loader",
-            description: "Loads the current target blueprint from the DB",
-            period: config.blueprints.period_secs_load,
-            task_impl: Box::new(blueprint_loader),
-            opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(rx_planner.clone())],
-            activator: task_blueprint_loader,
-        });
-
         // Background task: blueprint executor
         let blueprint_executor = blueprint_execution::BlueprintExecutor::new(
             datastore.clone(),
@@ -470,22 +444,6 @@ impl BackgroundTasksInitializer {
             activator: task_blueprint_executor,
         });
 
-        // Background task: CockroachDB node ID collector
-        let crdb_node_id_collector =
-            crdb_node_id_collector::CockroachNodeIdCollector::new(
-                datastore.clone(),
-                rx_blueprint.clone(),
-            );
-        driver.register(TaskDefinition {
-            name: "crdb_node_id_collector",
-            description: "Collects node IDs of running CockroachDB zones",
-            period: config.blueprints.period_secs_collect_crdb_node_ids,
-            task_impl: Box::new(crdb_node_id_collector),
-            opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(rx_blueprint.clone())],
-            activator: task_crdb_node_id_collector,
-        });
-
         // Background task: inventory collector
         //
         // This depends on the "output" of the blueprint executor in
@@ -515,6 +473,58 @@ impl BackgroundTasksInitializer {
             inventory_watcher
         };
 
+        // Background task: blueprint planner
+        //
+        // Replans on inventory collection and changes to the current
+        // target blueprint.
+        let blueprint_planner = blueprint_planner::BlueprintPlanner::new(
+            datastore.clone(),
+            false,
+            inventory_watcher.clone(),
+            rx_blueprint.clone(),
+        );
+        let rx_planner = blueprint_planner.watcher();
+        driver.register(TaskDefinition {
+            name: "blueprint_planner",
+            description: "Updates the target blueprint",
+            period: config.blueprints.period_secs_plan,
+            task_impl: Box::new(blueprint_planner),
+            opctx: opctx.child(BTreeMap::new()),
+            watchers: vec![
+                Box::new(inventory_watcher.clone()),
+                Box::new(rx_blueprint.clone()),
+            ],
+            activator: task_blueprint_planner,
+        });
+
+        // The loader watches the planner so that it can immediately load
+        // a new target blueprint.
+        driver.register(TaskDefinition {
+            name: "blueprint_loader",
+            description: "Loads the current target blueprint from the DB",
+            period: config.blueprints.period_secs_load,
+            task_impl: Box::new(blueprint_loader),
+            opctx: opctx.child(BTreeMap::new()),
+            watchers: vec![Box::new(rx_planner.clone())],
+            activator: task_blueprint_loader,
+        });
+
+        // Background task: CockroachDB node ID collector
+        let crdb_node_id_collector =
+            crdb_node_id_collector::CockroachNodeIdCollector::new(
+                datastore.clone(),
+                rx_blueprint.clone(),
+            );
+        driver.register(TaskDefinition {
+            name: "crdb_node_id_collector",
+            description: "Collects node IDs of running CockroachDB zones",
+            period: config.blueprints.period_secs_collect_crdb_node_ids,
+            task_impl: Box::new(crdb_node_id_collector),
+            opctx: opctx.child(BTreeMap::new()),
+            watchers: vec![Box::new(rx_blueprint.clone())],
+            activator: task_crdb_node_id_collector,
+        });
+
         // Cleans up and collects support bundles.
         //
         // This task is triggered by blueprint execution, since blueprint

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

@@ -13,6 +13,7 @@ use nexus_reconfigurator_planning::planner::Planner;
 use nexus_reconfigurator_preparation::PlanningInputFromDb;
 use nexus_types::deployment::{Blueprint, BlueprintTarget};
 use nexus_types::internal_api::background::BlueprintPlannerStatus;
+use omicron_uuid_kinds::CollectionUuid;
 use serde_json::json;
 use std::sync::Arc;
 use tokio::sync::watch::{self, Receiver, Sender};
@@ -21,6 +22,7 @@ use tokio::sync::watch::{self, Receiver, Sender};
 pub struct BlueprintPlanner {
     datastore: Arc<DataStore>,
     disabled: bool,
+    rx_inventory: Receiver<Option<CollectionUuid>>,
     rx_blueprint: Receiver<Option<Arc<(BlueprintTarget, Blueprint)>>>,
     tx_blueprint: Sender<Option<Arc<(BlueprintTarget, Blueprint)>>>,
 }
@@ -29,10 +31,11 @@ impl BlueprintPlanner {
     pub fn new(
         datastore: Arc<DataStore>,
         disabled: bool,
+        rx_inventory: Receiver<Option<CollectionUuid>>,
         rx_blueprint: Receiver<Option<Arc<(BlueprintTarget, Blueprint)>>>,
     ) -> Self {
         let (tx_blueprint, _) = watch::channel(None);
-        Self { datastore, disabled, rx_blueprint, tx_blueprint }
+        Self { datastore, disabled, rx_inventory, rx_blueprint, tx_blueprint }
     }
 
     pub fn watcher(
@@ -53,7 +56,8 @@ impl BlueprintPlanner {
         }
 
         // Get the current target blueprint to use as a parent.
-        let Some(rx) = self.rx_blueprint.borrow_and_update().clone() else {
+        // Cloned so that we don't block the channel.
+        let Some(loaded) = self.rx_blueprint.borrow_and_update().clone() else {
             warn!(
                 &opctx.log,
                 "blueprint planning skipped";
@@ -64,34 +68,54 @@ impl BlueprintPlanner {
             ));
             return status;
         };
-        let (target, parent) = &*rx;
+        let (target, parent) = &*loaded;
 
-        // Assemble the planning context.
-        let input = match PlanningInputFromDb::assemble(opctx, &self.datastore)
+        // Get the inventory most recently seen by the collection
+        // background task. The value is `Copy`, so with the deref
+        // we don't block the channel.
+        let Some(collection_id) = *self.rx_inventory.borrow_and_update() else {
+            warn!(
+                &opctx.log,
+                "blueprint planning skipped";
+                "reason" => "no inventory collection available"
+            );
+            status.error =
+                Some(String::from("no inventory collection available"));
+            return status;
+        };
+        let collection = match self
+            .datastore
+            .inventory_collection_read(opctx, collection_id)
             .await
         {
-            Ok(input) => input,
+            Ok(collection) => collection,
             Err(error) => {
-                error!(&opctx.log, "can't assemble planning input: {error}");
-                status.error = Some(error.to_string());
+                error!(
+                    &opctx.log,
+                    "can't read inventory collection";
+                    "collection_id" => %collection_id,
+                    "error" => %error,
+                );
+                status.error = Some(format!(
+                    "can't read inventory collection {}: {}",
+                    collection_id, error
+                ));
                 return status;
             }
         };
-        let inventory =
-            match self.datastore.inventory_get_latest_collection(opctx).await {
-                Ok(Some(inventory)) => inventory,
-                Ok(None) => {
-                    error!(&opctx.log, "no recent inventory collection");
-                    status.error =
-                        Some(String::from("no recent inventory collection"));
-                    return status;
-                }
+
+        // Assemble the planning context.
+        let input =
+            match PlanningInputFromDb::assemble(opctx, &self.datastore).await {
+                Ok(input) => input,
                 Err(error) => {
                     error!(
                         &opctx.log,
-                        "can't get latest inventory collection: {error}"
+                        "can't assemble planning input";
+                        "error" => %error,
                     );
-                    status.error = Some(error.to_string());
+                    status.error =
+                        Some(format!("can't assemble planning input: {error}"));
                     return status;
                 }
             };
@@ -102,7 +126,7 @@ impl BlueprintPlanner {
             &parent,
             &input,
             "blueprint_planner",
-            &inventory,
+            &collection,
         ) {
             Ok(planner) => planner,
             Err(error) => {
@@ -169,7 +193,7 @@ impl BlueprintPlanner {
             {
                 Ok(()) => (),
                 Err(error) => {
-                    error!(
+                    warn!(
                         &opctx.log,
                         "can't make blueprint the current target";
                         "error" => %error,