chore(bors): merge pull request #983

983: fix: allow scaling up of volumes of affinity group r=Abhinandan-Purkait a=Abhinandan-Purkait

### Issue
Volumes that are part of the affinity group consider restriction while scaling up for first time.

### Fix
The fix relaxes the restriction on first scale up.

Co-authored-by: Abhinandan Purkait <[email protected]>

Author: mayastor-bors

control-plane/agents/src/bin/core/controller/scheduling/affinity_group.rs

@@ -1,7 +1,7 @@
 use crate::controller::{registry::Registry, resources::ResourceUid};
 use std::collections::HashMap;
 use stor_port::types::v0::{
-    store::volume::{AffinityGroupSpec, VolumeSpec},
+    store::volume::{AffinityGroupSpec, VolumeOperation, VolumeSpec},
     transport::{NodeId, PoolId},
 };
 
@@ -12,32 +12,34 @@ pub(crate) fn get_restricted_nodes(
     affinity_group_spec: &AffinityGroupSpec,
     registry: &Registry,
 ) -> Vec<NodeId> {
-    let mut restricted_nodes: Vec<NodeId> = Vec::new();
-    if volume_spec.num_replicas == 1 {
-        let specs = registry.specs();
-        // Get the list of volumes being part of the Affinity Group.
-        let affinity_group_volumes = affinity_group_spec.volumes();
-        // Remove the current volume from the list, as it is under creation.
-        let affinity_group_volumes: Vec<_> = affinity_group_volumes
-            .iter()
-            .filter(|volume_id| *volume_id != volume_spec.uid())
-            .collect();
-        // List of restricted nodes, which already has replicas from the volumes of the
-        // Affinity Group.
-        for volume in &affinity_group_volumes {
-            // Fetch the list of nodes where the volume replicas are placed. If some
-            // volume doesn't exist yet, this list will be empty
-            // for that volume.
-            let node_ids = specs.volume_replica_nodes(volume);
-            // Add a new node in the list if it doesn't already exist.
-            let filtered_nodes: Vec<NodeId> = node_ids
-                .into_iter()
-                .filter(|id| !restricted_nodes.contains(id))
-                .collect();
-            restricted_nodes.extend(filtered_nodes);
-        }
+    // Since num_replicas in VolumeSpec is equal to 1 when scaling up from 1 replica to 2 replicas
+    // for the volume of an affinity group, if the number of nodes in cluster is equal to number of
+    // volumes in the affinity group we will exhaust nodes for scale up. This early exit relaxes the
+    // restriction for the first scale up (only support 1 replica add at a time).
+    if matches!(
+        volume_spec
+            .operation
+            .as_ref()
+            .map(|op| op.operation.clone()),
+        Some(VolumeOperation::SetReplica(2))
+    ) || volume_spec.num_replicas != 1
+    {
+        return Vec::new();
     }
-    restricted_nodes
+
+    let specs = registry.specs();
+    // List of restricted nodes, which already host replicas from the volumes of the affinity group
+    affinity_group_spec
+        .volumes()
+        .iter()
+        .filter(|&vid| vid != volume_spec.uid())
+        .flat_map(|vid| specs.volume_replica_nodes(vid))
+        .fold(Vec::new(), |mut acc, node_id| {
+            if !acc.contains(&node_id) {
+                acc.push(node_id);
+            }
+            acc
+        })
 }
 
 /// Get the map of pool to the number of the Affinity Group replica on the pool.

control-plane/agents/src/bin/core/tests/volume/affinity_group.rs

@@ -1,21 +1,24 @@
 use deployer_cluster::{Cluster, ClusterBuilder};
 use grpc::operations::{registry::traits::RegistryOperations, volume::traits::VolumeOperations};
-use stor_port::types::v0::transport::{AffinityGroup, CreateVolume, GetSpecs, VolumeId};
+use stor_port::types::v0::transport::{
+    AffinityGroup, CreateVolume, DestroyVolume, GetSpecs, SetVolumeReplica, VolumeId,
+};
 use tracing::info;
 
 #[tokio::test]
 async fn affinity_group() {
     let cluster = ClusterBuilder::builder()
         .with_rest(false)
         .with_agents(vec!["core"])
-        .with_io_engines(2)
-        .with_pools(2)
+        .with_io_engines(3)
+        .with_pools(3)
         .with_cache_period("1s")
         .build()
         .await
         .unwrap();
 
     startup_test(&cluster).await;
+    scale_up_test(&cluster).await;
 }
 
 async fn startup_test(cluster: &Cluster) {
@@ -31,7 +34,7 @@ async fn startup_test(cluster: &Cluster) {
 
     // Create all the volumes.
     for &item in &vols {
-        let _ = volume_client
+        volume_client
             .create(
                 &CreateVolume {
                     uuid: VolumeId::try_from(item.1).unwrap(),
@@ -42,7 +45,8 @@ async fn startup_test(cluster: &Cluster) {
                 },
                 None,
             )
-            .await;
+            .await
+            .expect("Volume creation should succeed");
     }
 
     // Restart the core-agent.
@@ -61,7 +65,7 @@ async fn startup_test(cluster: &Cluster) {
     let specs = registry_client
         .get_specs(&GetSpecs {}, None)
         .await
-        .expect("should be able to fetch specs");
+        .expect("Should be able to fetch specs");
 
     info!("Affinity Group Specs: {:?}", specs.affinity_groups);
 
@@ -86,4 +90,107 @@ async fn startup_test(cluster: &Cluster) {
             _ => {}
         }
     }
+
+    // Create all the volumes.
+    for &item in &vols {
+        volume_client
+            .destroy(
+                &DestroyVolume {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                },
+                None,
+            )
+            .await
+            .expect("Volume deletion should succeed");
+    }
+}
+
+async fn scale_up_test(cluster: &Cluster) {
+    let vols = vec![
+        (Some("ag1"), "eba487d9-0b57-407b-8b48-0b631a372183"),
+        (Some("ag1"), "359b7e1a-b724-443b-98b4-e6d97fabbb60"),
+        (Some("ag1"), "f2296d6a-77a6-401d-aad3-ccdc247b0a56"),
+    ];
+
+    let registry_client = cluster.grpc_client().registry();
+
+    // The Affinity Group specs should now have been loaded in memory.
+    // Fetch the specs.
+    let specs = registry_client
+        .get_specs(&GetSpecs {}, None)
+        .await
+        .expect("Should be able to fetch specs");
+
+    // Fail if there are affinity group specs from previous test lingering around.
+    assert_eq!(specs.affinity_groups.len(), 0);
+
+    let volume_client = cluster.grpc_client().volume();
+
+    for &item in &vols {
+        volume_client
+            .create(
+                &CreateVolume {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                    size: 5242880,
+                    replicas: 1,
+                    affinity_group: item.0.map(|val| AffinityGroup::new(val.to_string())),
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            .expect("Volume creation should succeed");
+    }
+
+    for &item in &vols {
+        volume_client
+            .set_replica(
+                &SetVolumeReplica {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                    replicas: 2,
+                },
+                None,
+            )
+            .await
+            .expect("Scale up should not fail");
+    }
+
+    for &item in &vols {
+        volume_client
+            .set_replica(
+                &SetVolumeReplica {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                    replicas: 3,
+                },
+                None,
+            )
+            .await
+            .expect("Scale up should not fail");
+    }
+
+    for &item in &vols {
+        volume_client
+            .set_replica(
+                &SetVolumeReplica {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                    replicas: 2,
+                },
+                None,
+            )
+            .await
+            .expect("Scale down should not fail");
+    }
+
+    for &item in &vols {
+        volume_client
+            .set_replica(
+                &SetVolumeReplica {
+                    uuid: VolumeId::try_from(item.1).unwrap(),
+                    replicas: 1,
+                },
+                None,
+            )
+            .await
+            .expect_err("Scale down should fail");
+    }
 }