[nexus] Add zpool to inventory (#5249)

This PR adds zpools to the inventory, and updates the existing
`omicron.public.zpool` table to only include aspects of the zpool which
are fully controlled by Nexus.

This PR is intended to solve a problem introduced by
#5172 . In that PR,
physical disk and zpool provisioning will be decided by Nexus, and sent
to the Sled Agent as a request. As a result, the `total_size` value of
zpools is not known when a pool is provisioned. Here, we add zpools to
the inventory, and JOIN with that table to access the "total_size" of
the zpool if it is known.

Additionally, this PR adds a test to validate that the database
(appropriately) throws "Insufficient capacity" errors if the Zpool
exists, but has not appeared in the inventory.

nexus/db-model/src/inventory.rs

@@ -9,7 +9,8 @@ use crate::schema::{
     hw_baseboard_id, inv_caboose, inv_collection, inv_collection_error,
     inv_omicron_zone, inv_omicron_zone_nic, inv_physical_disk,
     inv_root_of_trust, inv_root_of_trust_page, inv_service_processor,
-    inv_sled_agent, inv_sled_omicron_zones, sw_caboose, sw_root_of_trust_page,
+    inv_sled_agent, inv_sled_omicron_zones, inv_zpool, sw_caboose,
+    sw_root_of_trust_page,
 };
 use crate::PhysicalDiskKind;
 use crate::{
@@ -699,6 +700,43 @@ impl From<InvPhysicalDisk> for nexus_types::inventory::PhysicalDisk {
     }
 }
 
+/// See [`nexus_types::inventory::Zpool`].
+#[derive(Queryable, Clone, Debug, Selectable, Insertable)]
+#[diesel(table_name = inv_zpool)]
+pub struct InvZpool {
+    pub inv_collection_id: Uuid,
+    pub time_collected: DateTime<Utc>,
+    pub id: Uuid,
+    pub sled_id: Uuid,
+    pub total_size: ByteCount,
+}
+
+impl InvZpool {
+    pub fn new(
+        inv_collection_id: Uuid,
+        sled_id: Uuid,
+        zpool: &nexus_types::inventory::Zpool,
+    ) -> Self {
+        Self {
+            inv_collection_id,
+            time_collected: zpool.time_collected,
+            id: zpool.id,
+            sled_id,
+            total_size: zpool.total_size.into(),
+        }
+    }
+}
+
+impl From<InvZpool> for nexus_types::inventory::Zpool {
+    fn from(pool: InvZpool) -> Self {
+        Self {
+            time_collected: pool.time_collected,
+            id: pool.id,
+            total_size: *pool.total_size,
+        }
+    }
+}
+
 /// See [`nexus_types::inventory::OmicronZonesFound`].
 #[derive(Queryable, Clone, Debug, Selectable, Insertable)]
 #[diesel(table_name = inv_sled_omicron_zones)]

nexus/db-model/src/schema.rs

@@ -13,7 +13,7 @@ use omicron_common::api::external::SemverVersion;
 ///
 /// This should be updated whenever the schema is changed. For more details,
 /// refer to: schema/crdb/README.adoc
-pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(43, 0, 0);
+pub const SCHEMA_VERSION: SemverVersion = SemverVersion::new(44, 0, 0);
 
 table! {
     disk (id) {
@@ -960,8 +960,6 @@ table! {
 
         sled_id -> Uuid,
         physical_disk_id -> Uuid,
-
-        total_size -> Int8,
     }
 }
 
@@ -1375,6 +1373,16 @@ table! {
     }
 }
 
+table! {
+    inv_zpool (inv_collection_id, sled_id, id) {
+        inv_collection_id -> Uuid,
+        time_collected -> Timestamptz,
+        id -> Uuid,
+        sled_id -> Uuid,
+        total_size -> Int8,
+    }
+}
+
 table! {
     inv_sled_omicron_zones (inv_collection_id, sled_id) {
         inv_collection_id -> Uuid,

nexus/db-model/src/zpool.rs

@@ -2,7 +2,7 @@
 // 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/.
 
-use super::{ByteCount, Dataset, Generation};
+use super::{Dataset, Generation};
 use crate::collection::DatastoreCollectionConfig;
 use crate::schema::{dataset, zpool};
 use chrono::{DateTime, Utc};
@@ -26,26 +26,16 @@ pub struct Zpool {
 
     // The physical disk to which this Zpool is attached.
     pub physical_disk_id: Uuid,
-
-    // TODO: In the future, we may expand this structure to include
-    // size, allocation, and health information.
-    pub total_size: ByteCount,
 }
 
 impl Zpool {
-    pub fn new(
-        id: Uuid,
-        sled_id: Uuid,
-        physical_disk_id: Uuid,
-        total_size: ByteCount,
-    ) -> Self {
+    pub fn new(id: Uuid, sled_id: Uuid, physical_disk_id: Uuid) -> Self {
         Self {
             identity: ZpoolIdentity::new(id),
             time_deleted: None,
             rcgen: Generation::new(),
             sled_id,
             physical_disk_id,
-            total_size,
         }
     }
 }

nexus/db-queries/src/db/datastore/dataset.rs

@@ -229,12 +229,7 @@ mod test {
 
         // Create a fake zpool that backs our fake datasets.
         let zpool_id = Uuid::new_v4();
-        let zpool = Zpool::new(
-            zpool_id,
-            sled_id,
-            Uuid::new_v4(),
-            (1 << 30).try_into().unwrap(),
-        );
+        let zpool = Zpool::new(zpool_id, sled_id, Uuid::new_v4());
         datastore.zpool_upsert(zpool).await.expect("failed to upsert zpool");
 
         // Inserting a new dataset should succeed.

nexus/db-queries/src/db/datastore/inventory.rs

@@ -45,6 +45,7 @@ use nexus_db_model::InvRotPage;
 use nexus_db_model::InvServiceProcessor;
 use nexus_db_model::InvSledAgent;
 use nexus_db_model::InvSledOmicronZones;
+use nexus_db_model::InvZpool;
 use nexus_db_model::RotPageWhichEnum;
 use nexus_db_model::SledRole;
 use nexus_db_model::SledRoleEnum;
@@ -137,6 +138,18 @@ impl DataStore {
             })
             .collect();
 
+        // Pull zpools out of all sled agents
+        let zpools: Vec<_> = collection
+            .sled_agents
+            .iter()
+            .flat_map(|(sled_id, sled_agent)| {
+                sled_agent
+                    .zpools
+                    .iter()
+                    .map(|pool| InvZpool::new(collection_id, *sled_id, pool))
+            })
+            .collect();
+
         // Partition the sled agents into those with an associated baseboard id
         // and those without one.  We handle these pretty differently.
         let (sled_agents_baseboards, sled_agents_no_baseboards): (
@@ -670,6 +683,25 @@ impl DataStore {
                 }
             }
 
+            // Insert rows for all the zpools we found.
+            {
+                use db::schema::inv_zpool::dsl;
+
+                let batch_size = SQL_BATCH_SIZE.get().try_into().unwrap();
+                let mut zpools = zpools.into_iter();
+                loop {
+                    let some_zpools =
+                        zpools.by_ref().take(batch_size).collect::<Vec<_>>();
+                    if some_zpools.is_empty() {
+                        break;
+                    }
+                    let _ = diesel::insert_into(dsl::inv_zpool)
+                        .values(some_zpools)
+                        .execute_async(&conn)
+                        .await?;
+                }
+            }
+
             // Insert rows for the sled agents that we found.  In practice, we'd
             // expect these to all have baseboards (if using Oxide hardware) or
             // none have baseboards (if not).
@@ -1470,6 +1502,34 @@ impl DataStore {
             disks
         };
 
+        // Mapping of "Sled ID" -> "All zpools reported by that sled"
+        let zpools: BTreeMap<Uuid, Vec<nexus_types::inventory::Zpool>> = {
+            use db::schema::inv_zpool::dsl;
+
+            let mut zpools =
+                BTreeMap::<Uuid, Vec<nexus_types::inventory::Zpool>>::new();
+            let mut paginator = Paginator::new(batch_size);
+            while let Some(p) = paginator.next() {
+                let batch = paginated_multicolumn(
+                    dsl::inv_zpool,
+                    (dsl::sled_id, dsl::id),
+                    &p.current_pagparams(),
+                )
+                .filter(dsl::inv_collection_id.eq(id))
+                .select(InvZpool::as_select())
+                .load_async(&*conn)
+                .await
+                .map_err(|e| {
+                    public_error_from_diesel(e, ErrorHandler::Server)
+                })?;
+                paginator = p.found_batch(&batch, &|row| (row.sled_id, row.id));
+                for zpool in batch {
+                    zpools.entry(zpool.sled_id).or_default().push(zpool.into());
+                }
+            }
+            zpools
+        };
+
         // Collect the unique baseboard ids referenced by SPs, RoTs, and Sled
         // Agents.
         let baseboard_id_ids: BTreeSet<_> = sps
@@ -1577,6 +1637,10 @@ impl DataStore {
                             .get(&sled_id)
                             .map(|disks| disks.to_vec())
                             .unwrap_or_default(),
+                        zpools: zpools
+                            .get(&sled_id)
+                            .map(|zpools| zpools.to_vec())
+                            .unwrap_or_default(),
                     };
                     Ok((sled_id, sled_agent))
                 })