feat(snapshots, restores): store restores as a part of snap meta, exp…

…ose counts on rest

Signed-off-by: Abhinandan Purkait <[email protected]>

control-plane/agents/src/bin/core/controller/resources/operations.rs

@@ -1,5 +1,6 @@
-use crate::controller::registry::Registry;
+use crate::controller::{registry::Registry, resources::OperationGuardArc};
 use agents::errors::SvcError;
+use stor_port::types::v0::store::volume::VolumeSpec;
 
 /// Resource Cordon Operations.
 #[async_trait::async_trait]
@@ -265,11 +266,20 @@ pub(crate) trait ResourcePruning {
 pub(crate) trait ResourceCloning {
     type Create: Sync + Send;
     type CreateOutput: Sync + Send + Sized;
+    type Destroy: Sync + Send;
 
     /// Create a clone for the `Self` resource.
     async fn create_clone(
         &mut self,
         registry: &Registry,
         request: &Self::Create,
     ) -> Result<Self::CreateOutput, SvcError>;
+
+    /// Destroy a clone for the `Self` resource.
+    async fn destroy_clone(
+        &mut self,
+        registry: &Registry,
+        request: &Self::Destroy,
+        volume: OperationGuardArc<VolumeSpec>,
+    ) -> Result<(), SvcError>;
 }

control-plane/agents/src/bin/core/volume/clone_operations.rs

@@ -2,8 +2,8 @@ use crate::{
     controller::{
         registry::Registry,
         resources::{
-            operations::{ResourceCloning, ResourceLifecycleExt},
-            operations_helper::SpecOperationsHelper,
+            operations::{ResourceCloning, ResourceLifecycle, ResourceLifecycleExt},
+            operations_helper::{GuardedOperationsHelper, SpecOperationsHelper},
             OperationGuardArc, TraceStrLog,
         },
         scheduling::{volume::CloneVolumeSnapshot, ResourceFilter},
@@ -16,11 +16,13 @@ use stor_port::{
     types::v0::{
         store::{
             replica::ReplicaSpec,
-            snapshots::volume::VolumeSnapshot,
+            snapshots::volume::{
+                CreateRestoreInfo, DestroyRestoreInfo, VolumeSnapshot, VolumeSnapshotOperation,
+            },
             volume::{VolumeContentSource, VolumeSpec},
         },
         transport::{
-            CreateSnapshotVolume, Replica, SnapshotCloneId, SnapshotCloneParameters,
+            CreateSnapshotVolume, DestroyVolume, Replica, SnapshotCloneId, SnapshotCloneParameters,
             SnapshotCloneSpecParams,
         },
     },
@@ -45,14 +47,48 @@ impl SnapshotCloneOp<'_> {
 impl ResourceCloning for OperationGuardArc<VolumeSnapshot> {
     type Create = CreateSnapshotVolume;
     type CreateOutput = OperationGuardArc<VolumeSpec>;
+    type Destroy = DestroyVolume;
 
     async fn create_clone(
         &mut self,
         registry: &Registry,
         request: &Self::Create,
     ) -> Result<Self::CreateOutput, SvcError> {
+        let spec_clone = self
+            .start_update(
+                registry,
+                self.as_ref(),
+                VolumeSnapshotOperation::CreateRestore(CreateRestoreInfo::new(
+                    request.params().uuid.clone(),
+                )),
+            )
+            .await?;
         let request = CreateVolumeSource::Snapshot(SnapshotCloneOp(request, self));
-        OperationGuardArc::<VolumeSpec>::create_ext(registry, &request).await
+        let create_result = OperationGuardArc::<VolumeSpec>::create_ext(registry, &request).await;
+        self.complete_update(registry, create_result, spec_clone)
+            .await
+    }
+
+    async fn destroy_clone(
+        &mut self,
+        registry: &Registry,
+        request: &Self::Destroy,
+        mut volume: OperationGuardArc<VolumeSpec>,
+    ) -> Result<(), SvcError> {
+        let spec_clone = self
+            .start_update(
+                registry,
+                self.as_ref(),
+                VolumeSnapshotOperation::DestroyRestore(DestroyRestoreInfo::new(
+                    request.uuid.clone(),
+                )),
+            )
+            .await?;
+
+        let destroy_result = volume.destroy(registry, request).await;
+
+        self.complete_update(registry, destroy_result, spec_clone)
+            .await
     }
 }
 

control-plane/agents/src/bin/core/volume/operations.rs

@@ -43,7 +43,9 @@ use stor_port::{
     },
 };
 
+use crate::controller::resources::UpdateInnerValue;
 use std::{fmt::Debug, ops::Deref};
+use stor_port::types::v0::store::volume::VolumeContentSource;
 
 #[async_trait::async_trait]
 impl ResourceLifecycle for OperationGuardArc<VolumeSpec> {
@@ -174,6 +176,14 @@ impl ResourceLifecycle for OperationGuardArc<VolumeSpec> {
             }
         }
 
+        let spec = self.lock().clone();
+        if let Some(VolumeContentSource::Snapshot(snap_id, _)) = spec.content_source {
+            if let Ok(mut snap_guard) = specs.volume_snapshot(&snap_id).await {
+                snap_guard.lock().remove_restore(self.uuid());
+                snap_guard.update();
+            }
+        }
+
         self.complete_destroy(Ok(()), registry).await
     }
 }
@@ -791,6 +801,7 @@ impl ResourceLifecycleExt<CreateVolumeSource<'_>> for OperationGuardArc<VolumeSp
         volume
             .complete_create(result, registry, OnCreateFail::Delete)
             .await?;
+
         Ok(volume)
     }
 }

control-plane/agents/src/bin/core/volume/service.rs

@@ -30,14 +30,18 @@ use grpc::{
 use stor_port::{
     transport_api::{v0::Volumes, ReplyError, ResourceKind},
     types::v0::{
-        store::{snapshots::volume::VolumeSnapshotUserSpec, volume::VolumeSpec},
+        store::{
+            snapshots::volume::VolumeSnapshotUserSpec,
+            volume::{VolumeContentSource, VolumeSpec},
+        },
         transport::{
             CreateSnapshotVolume, CreateVolume, DestroyShutdownTargets, DestroyVolume, Filter,
             PublishVolume, RepublishVolume, SetVolumeReplica, ShareVolume, UnpublishVolume,
             UnshareVolume, Volume,
         },
     },
 };
+use tracing::info;
 
 #[derive(Debug, Clone)]
 pub(super) struct Service {
@@ -206,6 +210,7 @@ impl VolumeOperations for Service {
         let snapshots = self
             .get_snapshots(filter, ignore_notfound, pagination)
             .await?;
+        info!("{:?}", snapshots);
         Ok(snapshots)
     }
 
@@ -298,7 +303,28 @@ impl Service {
     #[tracing::instrument(level = "info", skip(self), err, fields(volume.uuid = %request.uuid))]
     pub(super) async fn destroy_volume(&self, request: &DestroyVolume) -> Result<(), SvcError> {
         let mut volume = self.specs().volume(&request.uuid).await?;
-        volume.destroy(&self.registry, request).await?;
+        let content_source = volume.as_ref().content_source.clone();
+        let snap_guard = match content_source {
+            None => None,
+            Some(VolumeContentSource::Snapshot(snap_uuid, _)) => {
+                match self.specs().volume_snapshot(&snap_uuid).await {
+                    Ok(snap_guard) => Some(snap_guard),
+                    Err(SvcError::VolSnapshotNotFound { .. }) => None,
+                    Err(error) => return Err(error),
+                }
+            }
+        };
+
+        match snap_guard {
+            None => {
+                volume.destroy(&self.registry, request).await?;
+            }
+            Some(mut snap_guard) => {
+                snap_guard
+                    .destroy_clone(&self.registry, request, volume)
+                    .await?;
+            }
+        }
         Ok(())
     }
 

control-plane/grpc/proto/v1/volume/volume.proto

@@ -59,6 +59,8 @@ message VolumeSpec {
   optional AffinityGroup affinity_group = 10;
   // Volume Content Source i.e the snapshot or a volume.
   optional VolumeContentSource content_source  = 11;
+  // Number of snapshots taken on this volume.
+  uint32 num_snapshots = 12;
 
   // Volume Content Source i.e the snapshot or a volume.
   message VolumeContentSource {
@@ -507,6 +509,8 @@ message VolumeSnapshotMeta {
   uint64                             spec_size = 6;
   // Size taken by the snapshot and its predecessors.
   uint64                  total_allocated_size = 7;
+  // Number of restores done from this snapshot.
+  uint32                           num_restore = 8;
 
   message ReplicaSnapshots {
     repeated ReplicaSnapshot         snapshots = 1;

control-plane/grpc/src/operations/volume/traits.rs

@@ -156,6 +156,7 @@ impl From<VolumeSpec> for volume::VolumeDefinition {
                 thin: volume_spec.thin,
                 affinity_group: volume_spec.affinity_group.into_opt(),
                 content_source: volume_spec.content_source.into_opt(),
+                num_snapshots: volume_spec.metadata.num_snapshots() as u32,
             }),
             metadata: Some(volume::Metadata {
                 spec_status: spec_status as i32,
@@ -318,6 +319,10 @@ impl TryFrom<volume::VolumeDefinition> for VolumeSpec {
             affinity_group: volume_spec.affinity_group.into_opt(),
             metadata: VolumeMetadata::new(volume_meta.as_thin),
             content_source: volume_spec.content_source.try_into_opt()?,
+            // Reviewer: Should we rather maintain the list of snapshots in the grpc layer and now
+            // return the count in rest api. Would that make the grpc layer slow if
+            // there are lots of snap or is the below good enough?
+            num_snapshots: volume_spec.num_snapshots as usize,
         };
         Ok(volume_spec)
     }

control-plane/grpc/src/operations/volume/traits_snapshots.rs

@@ -107,6 +107,7 @@ impl From<&stor_port::types::v0::store::snapshots::volume::VolumeSnapshot> for V
                 total_allocated_size: value.metadata().total_allocated_size(),
                 txn_id: value.metadata().txn_id().clone(),
                 transactions,
+                num_restores: value.num_restores(),
             },
         }
     }
@@ -137,6 +138,8 @@ pub struct VolumeSnapshotMeta {
     /// The "actual" snapshots can be accessed by the key `txn_id`.
     /// Failed transactions are any other key.
     transactions: HashMap<String, Vec<ReplicaSnapshot>>,
+    /// Volume metadata information.
+    num_restores: usize,
 }
 impl VolumeSnapshotMeta {
     /// Get the volume snapshot status.
@@ -167,6 +170,10 @@ impl VolumeSnapshotMeta {
     pub fn total_allocated_size(&self) -> u64 {
         self.total_allocated_size
     }
+    ///
+    pub fn num_restores(&self) -> usize {
+        self.num_restores
+    }
 }
 
 /// Volume replica snapshot information.
@@ -468,6 +475,7 @@ impl TryFrom<volume::VolumeSnapshot> for VolumeSnapshot {
                         snapshots.map(|s| (k, s))
                     })
                     .collect::<Result<HashMap<_, _>, _>>()?,
+                num_restores: meta.num_restore as usize,
             },
             state: VolumeSnapshotState {
                 info,
@@ -624,6 +632,7 @@ impl TryFrom<VolumeSnapshot> for volume::VolumeSnapshot {
                 size: value.meta.size,
                 spec_size: value.meta.spec_size,
                 total_allocated_size: value.meta.total_allocated_size,
+                num_restore: value.meta.num_restores as u32,
             }),
             state: Some(volume::VolumeSnapshotState {
                 state: Some(snapshot::SnapshotState {

control-plane/plugin/src/bin/rest-plugin/main.rs

@@ -2,8 +2,8 @@ use clap::Parser;
 use openapi::tower::client::Url;
 use plugin::{
     operations::{
-        Cordoning, Drain, Get, GetBlockDevices, GetSnapshots, List, Operations, RebuildHistory,
-        ReplicaTopology, Scale,
+        Cordoning, Drain, Get, GetBlockDevices, GetSnapshots, List, ListVolumes, Operations,
+        RebuildHistory, ReplicaTopology, Scale,
     },
     resources::{
         blockdevice, cordon, drain, node, pool, snapshot, volume, CordonResources, DrainResources,
@@ -83,13 +83,15 @@ async fn execute(cli_args: CliArgs) {
                 }
                 GetDrainArgs::Nodes => drain::NodeDrains::list(&cli_args.output).await,
             },
-            GetResources::Volumes => volume::Volumes::list(&cli_args.output).await,
+            GetResources::Volumes(vol_args) => {
+                volume::Volumes::list(&cli_args.output, vol_args).await
+            }
             GetResources::Volume { id } => volume::Volume::get(id, &cli_args.output).await,
             GetResources::RebuildHistory { id } => {
                 volume::Volume::rebuild_history(id, &cli_args.output).await
             }
-            GetResources::VolumeReplicaTopologies => {
-                volume::Volume::topologies(&cli_args.output).await
+            GetResources::VolumeReplicaTopologies(vol_args) => {
+                volume::Volume::topologies(&cli_args.output, vol_args).await
             }
             GetResources::VolumeReplicaTopology { id } => {
                 volume::Volume::topology(id, &cli_args.output).await

control-plane/plugin/src/operations.rs

@@ -1,4 +1,6 @@
-use crate::resources::{utils, CordonResources, DrainResources, GetResources, ScaleResources};
+use crate::resources::{
+    utils, volume::VolumesArgs, CordonResources, DrainResources, GetResources, ScaleResources,
+};
 use async_trait::async_trait;
 
 /// The types of operations that are supported.
@@ -41,6 +43,13 @@ pub trait List {
     async fn list(output: &utils::OutputFormat);
 }
 
+/// List trait.
+/// To be implemented by resources which support the 'list' operation.
+#[async_trait(?Send)]
+pub trait ListVolumes {
+    async fn list(output: &utils::OutputFormat, volume_args: &VolumesArgs);
+}
+
 /// Get trait.
 /// To be implemented by resources which support the 'get' operation.
 #[async_trait(?Send)]
@@ -62,7 +71,7 @@ pub trait Scale {
 #[async_trait(?Send)]
 pub trait ReplicaTopology {
     type ID;
-    async fn topologies(output: &utils::OutputFormat);
+    async fn topologies(output: &utils::OutputFormat, volume_args: &VolumesArgs);
     async fn topology(id: &Self::ID, output: &utils::OutputFormat);
 }
 

control-plane/plugin/src/resources/mod.rs

@@ -2,6 +2,7 @@ use crate::resources::{
     blockdevice::BlockDeviceArgs,
     node::{DrainNodeArgs, GetNodeArgs},
     snapshot::VolumeSnapshotArgs,
+    volume::VolumesArgs,
 };
 
 pub mod blockdevice;
@@ -29,13 +30,13 @@ pub enum GetResources {
     #[clap(subcommand)]
     Drain(GetDrainArgs),
     /// Get all volumes.
-    Volumes,
+    Volumes(VolumesArgs),
     /// Get volume with the given ID.
     Volume { id: VolumeId },
     /// Get Rebuild history for the volume with the given ID.
     RebuildHistory { id: VolumeId },
     /// Get the replica topology for all volumes.
-    VolumeReplicaTopologies,
+    VolumeReplicaTopologies(VolumesArgs),
     /// Get the replica topology for the volume with the given ID.
     VolumeReplicaTopology { id: VolumeId },
     /// Get volume snapshots based on input args.

control-plane/plugin/src/resources/snapshot.rs

@@ -56,7 +56,8 @@ impl CreateRow for openapi::models::VolumeSnapshot {
             ::utils::bytes::into_human(meta.spec_size),
             ::utils::bytes::into_human(state.allocated_size),
             ::utils::bytes::into_human(meta.total_allocated_size),
-            state.source_volume
+            state.source_volume,
+            self.definition.metadata.num_restores
         ]
     }
 }

control-plane/plugin/src/resources/utils.rs

@@ -20,15 +20,18 @@ lazy_static! {
         "STATUS",
         "SIZE",
         "THIN-PROVISIONED",
-        "ALLOCATED"
+        "ALLOCATED",
+        "SNAPSHOTS",
+        "SOURCE"
     ];
     pub static ref SNAPSHOT_HEADERS: Row = row![
         "ID",
         "TIMESTAMP",
         "SOURCE-SIZE",
         "ALLOCATED-SIZE",
         "TOTAL-ALLOCATED-SIZE",
-        "SOURCE-VOL"
+        "SOURCE-VOL",
+        "RESTORES"
     ];
     pub static ref POOLS_HEADERS: Row = row![
         "ID",