Cleanup Crucible resources with volume delete saga (#1706)

Add a volume delete saga, and call that whenever a volume needs to be
deleted. Crucible resources now have accounting for their use, and when
the number of references reaches zero then cleanup occurs.

Adds a table to store created region snapshots as part of this
accounting, and make a record of when a Crucible snapshot is taken of a
region. Region snapshots are uniquely identified by (dataset_id,
region_id, snapshot_id), and additionally contain a snapshot address for
identification as part of a volume construction request.

All DELETE calls to Crucible Agent(s) are now only performed by the
volume delete saga.

Add tests for cases where if Nexus is properly accounting for snapshots.
It should not delete the disk's corresponding region when the disk is
deleted because a snapshot exists, or if multiple disks reference a
running snapshot.

A small change to the simulated Crucible agent is also included in this
commit - logic was moved out of the HTTP endpoint function into the
Crucible struct. Logic doesn't belong in HTTP endpoint functions :)

Closes #1632

Author: jmpesp

common/src/sql/dbinit.sql

@@ -229,6 +229,45 @@ CREATE INDEX on omicron.public.Region (
     dataset_id
 );
 
+/*
+ * A snapshot of a region, within a dataset.
+ */
+CREATE TABLE omicron.public.region_snapshot (
+    dataset_id UUID NOT NULL,
+    region_id UUID NOT NULL,
+
+    /* Associated higher level virtual snapshot */
+    snapshot_id UUID NOT NULL,
+
+    /*
+     * Target string, for identification as part of
+     * volume construction request(s)
+     */
+    snapshot_addr TEXT NOT NULL,
+
+    /* How many volumes reference this? */
+    volume_references INT8 NOT NULL,
+
+    PRIMARY KEY (dataset_id, region_id, snapshot_id)
+);
+
+/* Index for use during join with region table */
+CREATE INDEX on omicron.public.region_snapshot (
+    dataset_id, region_id
+);
+
+/*
+ * Index on volume_references and snapshot_addr for crucible
+ * resource accounting lookup
+ */
+CREATE INDEX on omicron.public.region_snapshot (
+    volume_references
+);
+
+CREATE INDEX on omicron.public.region_snapshot (
+    snapshot_addr
+);
+
 /*
  * A volume within Crucible
  */
@@ -247,7 +286,19 @@ CREATE TABLE omicron.public.volume (
      * consumed by some Upstairs code to perform the volume creation. The Rust
      * type of this column should be Crucible::VolumeConstructionRequest.
      */
-    data TEXT NOT NULL
+    data TEXT NOT NULL,
+
+    /*
+     * A JSON document describing what resources to clean up when deleting this
+     * volume. The Rust type of this column should be the CrucibleResources
+     * enum.
+     */
+    resources_to_clean_up TEXT
+);
+
+/* Quickly find deleted volumes */
+CREATE INDEX on omicron.public.volume (
+    time_deleted
 );
 
 /*

nexus/db-model/src/lib.rs

@@ -45,6 +45,7 @@ mod project;
 pub mod queries;
 mod rack;
 mod region;
+mod region_snapshot;
 mod role_assignment;
 mod role_builtin;
 pub mod saga_types;
@@ -110,6 +111,7 @@ pub use producer_endpoint::*;
 pub use project::*;
 pub use rack::*;
 pub use region::*;
+pub use region_snapshot::*;
 pub use role_assignment::*;
 pub use role_builtin::*;
 pub use service::*;

nexus/db-model/src/region_snapshot.rs

@@ -0,0 +1,35 @@
+// 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/.
+
+use crate::schema::region_snapshot;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+/// Database representation of a Region's snapshot.
+///
+/// A region snapshot represents a snapshot of a region, taken during the higher
+/// level virtual disk snapshot operation.
+#[derive(
+    Queryable,
+    Insertable,
+    Debug,
+    Clone,
+    Selectable,
+    Serialize,
+    Deserialize,
+    PartialEq,
+)]
+#[diesel(table_name = region_snapshot)]
+pub struct RegionSnapshot {
+    // unique identifier of this region snapshot
+    pub dataset_id: Uuid,
+    pub region_id: Uuid,
+    pub snapshot_id: Uuid,
+
+    // used for identifying volumes that reference this
+    pub snapshot_addr: String,
+
+    // how many volumes reference this?
+    pub volume_references: i64,
+}

nexus/db-model/src/schema.rs

@@ -441,6 +441,16 @@ table! {
     }
 }
 
+table! {
+    region_snapshot (dataset_id, region_id, snapshot_id) {
+        dataset_id -> Uuid,
+        region_id -> Uuid,
+        snapshot_id -> Uuid,
+        snapshot_addr -> Text,
+        volume_references -> Int8,
+    }
+}
+
 table! {
     volume (id) {
         id -> Uuid,
@@ -450,7 +460,12 @@ table! {
         rcgen -> Int8,
 
         data -> Text,
-        /* TODO: some sort of refcount? */
+
+        /*
+         * During volume deletion, a serialized list of Crucible resources to
+         * clean up will be written here, along with setting time_deleted.
+         */
+        resources_to_clean_up -> Nullable<Text>,
     }
 }
 
@@ -622,6 +637,7 @@ allow_tables_to_appear_in_same_query!(
     project,
     rack,
     region,
+    region_snapshot,
     saga,
     saga_node_event,
     silo,
@@ -630,6 +646,7 @@ allow_tables_to_appear_in_same_query!(
     service,
     sled,
     router_route,
+    volume,
     vpc,
     vpc_subnet,
     vpc_router,

nexus/db-model/src/volume.rs

@@ -24,11 +24,13 @@ use uuid::Uuid;
 pub struct Volume {
     #[diesel(embed)]
     identity: VolumeIdentity,
-    time_deleted: Option<DateTime<Utc>>,
+    pub time_deleted: Option<DateTime<Utc>>,
 
     rcgen: Generation,
 
     data: String,
+
+    pub resources_to_clean_up: Option<String>,
 }
 
 impl Volume {
@@ -38,6 +40,7 @@ impl Volume {
             time_deleted: None,
             rcgen: Generation::new(),
             data,
+            resources_to_clean_up: None,
         }
     }
 

nexus/src/app/disk.rs

@@ -510,6 +510,9 @@ impl super::Nexus {
             .project_delete_snapshot(opctx, &authz_snapshot, &db_snapshot)
             .await?;
 
+        // Kick off volume deletion saga
+        self.volume_delete(db_snapshot.volume_id).await?;
+
         Ok(())
     }
 }

nexus/src/app/mod.rs

@@ -39,6 +39,7 @@ mod silo;
 mod sled;
 pub mod test_interfaces;
 mod update;
+mod volume;
 mod vpc;
 mod vpc_router;
 mod vpc_subnet;

nexus/src/app/sagas/disk_create.rs

@@ -4,7 +4,7 @@
 
 use super::{
     ActionRegistry, NexusActionContext, NexusSaga, SagaInitError,
-    ACTION_GENERATE_ID,
+    ACTION_GENERATE_ID, MAX_CONCURRENT_REGION_REQUESTS,
 };
 use crate::app::sagas::NexusAction;
 use crate::context::OpContext;
@@ -51,16 +51,10 @@ lazy_static! {
         sdc_create_disk_record,
         sdc_create_disk_record_undo
     );
-    static ref REGIONS_ALLOC: NexusAction = ActionFunc::new_action(
-        "disk-create.regions-alloc",
-        sdc_alloc_regions,
-        sdc_alloc_regions_undo
-    );
-    static ref REGIONS_ENSURE: NexusAction = ActionFunc::new_action(
-        "disk-create.regions-ensure",
-        sdc_regions_ensure,
-        sdc_regions_ensure_undo
-    );
+    static ref REGIONS_ALLOC: NexusAction =
+        new_action_noop_undo("disk-create.regions-alloc", sdc_alloc_regions,);
+    static ref REGIONS_ENSURE: NexusAction =
+        new_action_noop_undo("disk-create.regions-ensure", sdc_regions_ensure,);
     static ref CREATE_VOLUME_RECORD: NexusAction = ActionFunc::new_action(
         "disk-create.create-volume-record",
         sdc_create_volume_record,
@@ -231,6 +225,7 @@ async fn sdc_alloc_regions(
     let osagactx = sagactx.user_data();
     let params = sagactx.saga_params::<Params>()?;
     let volume_id = sagactx.lookup::<Uuid>("volume_id")?;
+
     // Ensure the disk is backed by appropriate regions.
     //
     // This allocates regions in the database, but the disk state is still
@@ -251,16 +246,7 @@ async fn sdc_alloc_regions(
     Ok(datasets_and_regions)
 }
 
-async fn sdc_alloc_regions_undo(
-    sagactx: NexusActionContext,
-) -> Result<(), anyhow::Error> {
-    let osagactx = sagactx.user_data();
-
-    let volume_id = sagactx.lookup::<Uuid>("volume_id")?;
-    osagactx.datastore().regions_hard_delete(volume_id).await?;
-    Ok(())
-}
-
+/// Call out to Crucible agent and perform region creation.
 async fn ensure_region_in_dataset(
     log: &Logger,
     dataset: &db::model::Dataset,
@@ -317,10 +303,7 @@ async fn ensure_region_in_dataset(
     Ok(region.into_inner())
 }
 
-// Arbitrary limit on concurrency, for operations issued
-// on multiple regions within a disk at the same time.
-const MAX_CONCURRENT_REGION_REQUESTS: usize = 3;
-
+/// Call out to Crucible agent and perform region creation.
 async fn sdc_regions_ensure(
     sagactx: NexusActionContext,
 ) -> Result<String, ActionError> {
@@ -379,7 +362,7 @@ async fn sdc_regions_ensure(
 
     let block_size = datasets_and_regions[0].1.block_size;
 
-    // If requested, back disk by image
+    // If a disk source was requested, set the read-only parent of this disk.
     let osagactx = sagactx.user_data();
     let params = sagactx.saga_params::<Params>()?;
     let log = osagactx.log();
@@ -497,7 +480,7 @@ async fn sdc_regions_ensure(
         );
     }
 
-    // Store volume details in db
+    // Create volume construction request for this disk
     let mut rng = StdRng::from_entropy();
     let volume_construction_request = VolumeConstructionRequest::Volume {
         id: disk_id,
@@ -554,55 +537,6 @@ async fn sdc_regions_ensure(
     Ok(volume_data)
 }
 
-pub(super) async fn delete_regions(
-    datasets_and_regions: Vec<(db::model::Dataset, db::model::Region)>,
-) -> Result<(), Error> {
-    let request_count = datasets_and_regions.len();
-    futures::stream::iter(datasets_and_regions)
-        .map(|(dataset, region)| async move {
-            let url = format!("http://{}", dataset.address());
-            let client = CrucibleAgentClient::new(&url);
-            let id = RegionId(region.id().to_string());
-            client.region_delete(&id).await.map_err(|e| match e {
-                crucible_agent_client::Error::ErrorResponse(rv) => {
-                    match rv.status() {
-                        http::StatusCode::SERVICE_UNAVAILABLE => {
-                            Error::unavail(&rv.message)
-                        }
-                        status if status.is_client_error() => {
-                            Error::invalid_request(&rv.message)
-                        }
-                        _ => Error::internal_error(&rv.message),
-                    }
-                }
-                _ => Error::internal_error(
-                    "unexpected failure during `region_delete`",
-                ),
-            })
-        })
-        // Execute the allocation requests concurrently.
-        .buffer_unordered(std::cmp::min(
-            request_count,
-            MAX_CONCURRENT_REGION_REQUESTS,
-        ))
-        .collect::<Vec<Result<_, _>>>()
-        .await
-        .into_iter()
-        .collect::<Result<Vec<_>, _>>()?;
-    Ok(())
-}
-
-async fn sdc_regions_ensure_undo(
-    sagactx: NexusActionContext,
-) -> Result<(), anyhow::Error> {
-    let datasets_and_regions = sagactx
-        .lookup::<Vec<(db::model::Dataset, db::model::Region)>>(
-            "datasets_and_regions",
-        )?;
-    delete_regions(datasets_and_regions).await?;
-    Ok(())
-}
-
 async fn sdc_create_volume_record(
     sagactx: NexusActionContext,
 ) -> Result<db::model::Volume, ActionError> {
@@ -628,7 +562,7 @@ async fn sdc_create_volume_record_undo(
     let osagactx = sagactx.user_data();
 
     let volume_id = sagactx.lookup::<Uuid>("volume_id")?;
-    osagactx.datastore().volume_delete(volume_id).await?;
+    osagactx.nexus().volume_delete(volume_id).await?;
     Ok(())
 }
 
@@ -647,6 +581,7 @@ async fn sdc_finalize_disk_record(
         .lookup_for(authz::Action::Modify)
         .await
         .map_err(ActionError::action_failed)?;
+
     // TODO-security Review whether this can ever fail an authz check.  We don't
     // want this to ever fail the authz check here -- if it did, we would have
     // wanted to catch that a lot sooner.  It wouldn't make sense for it to fail
@@ -665,6 +600,7 @@ async fn sdc_finalize_disk_record(
         )
         .await
         .map_err(ActionError::action_failed)?;
+
     Ok(())
 }