fix(csi-controller/gc): list volumes with pagination

Signed-off-by: Tiago Castro <[email protected]>
openebs · Jan 24, 2024 · e4a5f56 · e4a5f56
1 parent 4576447
commit e4a5f56
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 14 deletions.
diff --git a/control-plane/csi-driver/src/bin/controller/client.rs b/control-plane/csi-driver/src/bin/controller/client.rs
@@ -148,6 +148,12 @@ impl IoEngineApiClient {
     }
 }
 
+/// Token used to list volumes with pagination.
+pub(crate) enum ListToken {
+    String(String),
+    Number(isize),
+}
+
 impl IoEngineApiClient {
     /// List all nodes available in IoEngine cluster.
     pub(crate) async fn list_nodes(&self) -> Result<Vec<Node>, ApiClientError> {
@@ -178,17 +184,17 @@ impl IoEngineApiClient {
     pub(crate) async fn list_volumes(
         &self,
         max_entries: i32,
-        starting_token: String,
+        starting_token: ListToken,
     ) -> Result<Volumes, ApiClientError> {
         let max_entries = max_entries as isize;
-        let starting_token = if starting_token.is_empty() {
-            0
-        } else {
-            starting_token.parse::<isize>().map_err(|_| {
+        let starting_token = match starting_token {
+            ListToken::String(starting_token) if starting_token.is_empty() => 0,
+            ListToken::String(starting_token) => starting_token.parse::<isize>().map_err(|_| {
                 ApiClientError::InvalidArgument(
                     "Failed to parse starting token as an isize".to_string(),
                 )
-            })?
+            })?,
+            ListToken::Number(starting_token) => starting_token,
         };
 
         let response = self

diff --git a/control-plane/csi-driver/src/bin/controller/controller.rs b/control-plane/csi-driver/src/bin/controller/controller.rs
@@ -1,4 +1,6 @@
-use crate::{ApiClientError, CreateVolumeTopology, CsiControllerConfig, IoEngineApiClient};
+use crate::{
+    client::ListToken, ApiClientError, CreateVolumeTopology, CsiControllerConfig, IoEngineApiClient,
+};
 
 use csi_driver::context::{CreateParams, PublishParams};
 use rpc::csi::{volume_content_source::Type, Topology as CsiTopology, *};
@@ -652,7 +654,7 @@ impl rpc::csi::controller_server::Controller for CsiControllerSvc {
         let vt_mapper = VolumeTopologyMapper::init().await?;
 
         let volumes = IoEngineApiClient::get_client()
-            .list_volumes(max_entries, args.starting_token)
+            .list_volumes(max_entries, ListToken::String(args.starting_token))
             .await
             .map_err(|e| Status::internal(format!("Failed to list volumes, error = {e:?}")))?;
 

diff --git a/control-plane/csi-driver/src/bin/controller/pvwatcher.rs b/control-plane/csi-driver/src/bin/controller/pvwatcher.rs
@@ -1,3 +1,4 @@
+use crate::client::ListToken;
 use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::PersistentVolume;
 use kube::{
@@ -100,15 +101,27 @@ impl PvGarbageCollector {
     /// 2. to tackle k8s bug where volumes are leaked when PV deletion is attempted before
     /// PVC deletion.
     async fn delete_orphan_volumes(&self) {
-        match self.rest_client.list_volumes(0, "".to_string()).await {
-            Ok(volume_list) => {
-                for vol in volume_list.entries {
-                    if self.is_vol_orphaned(&vol.spec.uuid).await {
-                        self.delete_volume(vol.spec.uuid).await;
+        let max_entries = 200;
+        let mut starting_token = Some(0);
+        while let Some(token) = starting_token {
+            match self
+                .rest_client
+                .list_volumes(max_entries, ListToken::Number(token))
+                .await
+            {
+                Ok(volumes) => {
+                    starting_token = volumes.next_token;
+                    for vol in volumes.entries {
+                        if self.is_vol_orphaned(&vol.spec.uuid).await {
+                            self.delete_volume(vol.spec.uuid).await;
+                        }
                     }
                 }
+                Err(error) => {
+                    error!(?error, "Unable to list volumes");
+                    return;
+                }
             }
-            Err(error) => error!(?error, "Unable to list volumes"),
         }
     }
 

diff --git a/control-plane/csi-driver/src/bin/controller/server.rs b/control-plane/csi-driver/src/bin/controller/server.rs
@@ -33,6 +33,7 @@ impl Connected for UnixStream {
     }
 }
 
+// Not sure why we need the inner fields, probably worth checking if we can remove them.
 #[derive(Clone, Debug)]
 #[allow(unused)]
 struct UdsConnectInfo {