chore(bors): merge pull request #1038

1038: feat(affinity-group): support scale down to 1 replica (backport #1037) r=tiagolobocastro a=mergify[bot]

Scaling down affinity groups to 1 replica was previously blocked because we didn't want to get into a situation where replicas would end up sharing the same node, defeating the purpose of Affinity Groups!

We now loosen this restriction by allowing the scale down if we don't get into that scenario :)

This is done by creating a list of restricted nodes which are nodes where a 1-replica volume part of the same affinity group places its replica. In such case, we don't want to scale down in such a way that the last remaining replica would be placed in restricted nodes!

This isn't the full monty as replicas may be sharing nodes when Scaling back from 3 to 2, which would end up impeding further scaling down to 1! For this we'd have to implement some logic to improve the sorting. This limitation is fine for now, we can document scaling up and down as a way of moving replicas around!<hr>This is an automatic backport of pull request #1037 done by [Mergify](https://mergify.com).

Co-authored-by: Tiago Castro <[email protected]>

Author: mayastor-bors

.github/mergify.yml

@@ -0,0 +1,13 @@
+pull_request_rules:
+  - name: auto-label and approve backport PRs
+    conditions:
+      - author=mergify[bot]
+      - title~=\(backport \#[0-9]+\)$
+    actions:
+      label:
+        add:
+          - kind/backport
+      review:
+        type: APPROVE
+        message: "Automatically approved backport PR"
+        bot_account: openebs-ci

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

@@ -1,5 +1,5 @@
 use crate::controller::{registry::Registry, resources::ResourceUid};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use stor_port::types::v0::{
     store::volume::{AffinityGroupSpec, VolumeOperation, VolumeSpec},
     transport::{NodeId, PoolId},
@@ -61,6 +61,27 @@ pub(crate) async fn get_pool_ag_replica_count(
     pool_ag_replica_count
 }
 
+/// Get a set of nodes which contain the replica of a single-replica volume, part of the affinity group.
+pub(crate) async fn ag_restricted_nodes(
+    affinity_group_spec: &AffinityGroupSpec,
+    registry: &Registry,
+) -> HashSet<NodeId> {
+    let mut nodes = HashSet::<NodeId>::new();
+    let specs = registry.specs();
+    for volume_id in affinity_group_spec.volumes() {
+        let Some(volume) = registry.specs().volume_rsc(volume_id) else {
+            // should not really happen
+            continue;
+        };
+        if volume.lock().num_replicas != 1 {
+            continue;
+        }
+
+        nodes.extend(specs.volume_replica_nodes(volume_id));
+    }
+    nodes
+}
+
 /// Get the map of node to the number of the Affinity Group nexuses on the node.
 pub(crate) async fn get_node_ag_nexus_count(
     affinity_group_spec: &AffinityGroupSpec,

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

@@ -332,13 +332,23 @@ impl ChildSorters {
     /// Sort replicas by their nexus child (state and rebuild progress)
     /// todo: should we use weights instead (like moac)?
     pub(crate) fn sort(
-        _request: &GetChildForRemovalContext,
+        request: &GetChildForRemovalContext,
         a: &ReplicaItem,
         b: &ReplicaItem,
     ) -> std::cmp::Ordering {
         match Self::sort_by_health(a, b) {
             Ordering::Equal => match Self::sort_by_child(a, b) {
                 Ordering::Equal => {
+                    if let Some(ag) = request.affinity_group() {
+                        match (
+                            a.ag_restricted_node(ag.restricted_nodes()),
+                            b.ag_restricted_node(ag.restricted_nodes()),
+                        ) {
+                            (Some(a), Some(b)) if a != b => return b.cmp(&a),
+                            _ => {}
+                        }
+                    }
+
                     // Remove replicas from nodes which are cordoned with most priority.
                     // remove mismatched topology replicas first
                     if let (Some(a), Some(b)) = (a.valid_node_topology(), b.valid_node_topology()) {
@@ -377,6 +387,7 @@ impl ChildSorters {
                     let childb_is_local = !b.spec().share.shared();
                     match (childa_is_local, childb_is_local) {
                         (true, true) | (false, false) => {
+                            // todo: this should probably be done regardless of child locality
                             b.ag_replicas_on_pool().cmp(&a.ag_replicas_on_pool())
                         }
                         (true, false) => Ordering::Greater,

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

@@ -16,7 +16,11 @@ use stor_port::types::v0::{
     transport::{Child, ChildUri, NodeId, PoolId, Replica},
 };
 
-use std::{cmp::Ordering, collections::HashMap, ops::Deref};
+use std::{
+    cmp::Ordering,
+    collections::{HashMap, HashSet},
+    ops::Deref,
+};
 
 /// Item for pool scheduling logic.
 #[derive(Debug, Clone)]
@@ -293,6 +297,10 @@ impl ReplicaItem {
         self.valid_pool_topology = valid_pool_topology;
         self
     }
+    /// Check all topological information is valid!
+    pub(crate) fn valid_topology(&self) -> bool {
+        self.valid_pool_topology() == &Some(true) && self.valid_node_topology() == Some(true)
+    }
     /// Get a reference to the node topology validity information.
     pub(crate) fn node_topology_info(&self) -> &Option<TopologyRmInfo> {
         &self.node_topology_info
@@ -338,6 +346,13 @@ impl ReplicaItem {
     pub(crate) fn ag_replicas_on_pool(&self) -> u64 {
         self.ag_replicas_on_pool.unwrap_or(u64::MIN)
     }
+    /// Check if this replica is in an Affinity Group restricted node.
+    /// A restricted node is a node where another single-replica volume (from the same AG) already
+    /// resides in.
+    pub(crate) fn ag_restricted_node(&self, restricted_nodes: &HashSet<NodeId>) -> Option<bool> {
+        let node_spec = self.node_spec.as_ref()?;
+        Some(restricted_nodes.get(node_spec.id()).is_some())
+    }
 }
 
 /// Individual nexus child (replicas) which can be used for nexus creation.

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

@@ -22,11 +22,12 @@ use stor_port::types::v0::{
     transport::{NodeId, PoolId, VolumeState},
 };
 
-use crate::controller::scheduling::ResourceExcReason;
+use crate::controller::scheduling::{affinity_group::ag_restricted_nodes, ResourceExcReason};
 use std::{
-    collections::{BTreeMap, HashMap},
+    collections::{BTreeMap, HashMap, HashSet},
     ops::Deref,
 };
+use stor_port::types::v0::store::volume::AffinityGroupSpec;
 
 /// Move replica to another pool.
 #[derive(Default, Clone)]
@@ -273,6 +274,20 @@ impl GetChildForRemoval {
     }
 }
 
+/// Affinity Group useful information when scaling down.
+#[derive(Clone)]
+pub(crate) struct AffinityGroupRmCtx {
+    _group: AffinityGroupSpec,
+    pool_repl_cnt: HashMap<PoolId, u64>,
+    restricted_nodes: HashSet<NodeId>,
+}
+impl AffinityGroupRmCtx {
+    /// Get the Affinity Group restricted nodes.
+    pub(crate) fn restricted_nodes(&self) -> &HashSet<NodeId> {
+        &self.restricted_nodes
+    }
+}
+
 /// Used to filter nexus children in order to choose the best candidates for removal
 /// when the volume's replica count is being reduced.
 #[derive(Clone)]
@@ -282,6 +297,7 @@ pub(crate) struct GetChildForRemovalContext {
     state: VolumeState,
     nexus_info: Option<NexusInfo>,
     unused_only: bool,
+    affinity_group: Option<AffinityGroupRmCtx>,
 }
 impl std::fmt::Debug for GetChildForRemovalContext {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -295,7 +311,11 @@ impl std::fmt::Debug for GetChildForRemovalContext {
 }
 
 impl GetChildForRemovalContext {
-    async fn new(registry: &Registry, request: &GetChildForRemoval) -> Result<Self, SvcError> {
+    async fn new(
+        registry: &Registry,
+        request: &GetChildForRemoval,
+        affinity_group: Option<AffinityGroupSpec>,
+    ) -> Result<Self, SvcError> {
         let nexus_info = registry
             .nexus_info(
                 Some(&request.spec.uuid),
@@ -311,10 +331,19 @@ impl GetChildForRemovalContext {
             state: request.state.clone(),
             nexus_info,
             unused_only: request.unused_only,
+            affinity_group: match affinity_group {
+                None => None,
+                Some(group) => Some(AffinityGroupRmCtx {
+                    pool_repl_cnt: get_pool_ag_replica_count(&group, registry).await,
+                    restricted_nodes: ag_restricted_nodes(&group, registry).await,
+                    _group: group,
+                }),
+            },
         })
     }
 
-    async fn list(&self, pool_ag_rep: &Option<HashMap<PoolId, u64>>) -> Vec<ReplicaItem> {
+    async fn list(&self) -> Vec<ReplicaItem> {
+        let pool_ag_rep = self.affinity_group.as_ref().map(|g| &g.pool_repl_cnt);
         let replicas = self.registry.specs().volume_replicas_cln(&self.spec.uuid);
         let nexus = self.registry.specs().volume_target_nexus_rsc(&self.spec);
         let used_node_spread =
@@ -386,19 +415,22 @@ impl GetChildForRemovalContext {
             })
             .collect::<Vec<_>>()
     }
+
+    /// Get the volume's affinity group.
+    pub(crate) fn affinity_group(&self) -> Option<&AffinityGroupRmCtx> {
+        self.affinity_group.as_ref()
+    }
 }
 
 impl DecreaseVolumeReplica {
     async fn builder(request: &GetChildForRemoval, registry: &Registry) -> Result<Self, SvcError> {
-        let mut pool_ag_replica_count_map: Option<HashMap<PoolId, u64>> = None;
-        if let Some(affinity_group) = &request.spec.affinity_group {
-            let affinity_group_spec = registry.specs().affinity_group_spec(affinity_group.id())?;
-            pool_ag_replica_count_map =
-                Some(get_pool_ag_replica_count(&affinity_group_spec, registry).await);
-        }
+        let affinity_group = request.spec.affinity_group.as_ref();
+        let affinity_group = affinity_group
+            .map(|g| registry.specs().affinity_group_spec(g.id()))
+            .transpose()?;
 
-        let context = GetChildForRemovalContext::new(registry, request).await?;
-        let list = context.list(&pool_ag_replica_count_map).await;
+        let context = GetChildForRemovalContext::new(registry, request, affinity_group).await?;
+        let list = context.list().await;
         Ok(Self {
             data: ResourceData::new(context, list),
         })
@@ -448,6 +480,24 @@ impl ReplicaRemovalCandidates {
             None
         }
     }
+    fn ag_restricted_nodes(&self) -> Option<&HashSet<NodeId>> {
+        let ag = self.context.affinity_group.as_ref()?;
+        if ag.restricted_nodes.is_empty() {
+            return None;
+        }
+        //     N
+        // R 1 2 3
+        // 1 x x
+        // 2 x x
+        // 3 x x x
+        // todo: In this case, shouldn't we block removal or r3n3 ?
+        if self.context.spec.num_replicas > 2 {
+            // for now, we only restrict from 2->1
+            return None;
+        }
+        Some(&ag.restricted_nodes)
+    }
+
     /// Get the next unhealthy candidates (any is a good fit).
     fn next_unhealthy(&mut self) -> Option<ReplicaItem> {
         self.unhealthy.pop()
@@ -457,6 +507,46 @@ impl ReplicaRemovalCandidates {
     pub(crate) fn next(&mut self) -> Option<ReplicaItem> {
         self.next_unhealthy().or_else(|| self.next_healthy())
     }
+    /// Get the next removal candidate.
+    /// Unhealthy replicas are removed before healthy replicas.
+    /// Use when scaling down the volume where we need to guard against restricted node usage.
+    pub(crate) fn next_down(&mut self) -> Option<Option<ReplicaItem>> {
+        // we don't want to leave place 1-r affinity volumes in the same node (potentially),
+        // so we should check whether we have other replicas available
+        let mut replicas_in_unrestricted_nodes = false;
+
+        let candidate = self.next_unhealthy().or_else(|| self.next_healthy())?;
+
+        let Some(restricted_nodes) = self.ag_restricted_nodes() else {
+            // no affinity groups, just remove the best candidate
+            return Some(Some(candidate));
+        };
+        // todo: still need a better way of controlling the priority of candidates when
+        //  scaling from X -> X-1, where X > 2
+        //  also we can block scale down to 1, if we don't have sufficient distinct nodes
+        //  to place all the other volume replicas
+
+        tracing::debug!("Restricted Affinity Group Nodes: {restricted_nodes:?}");
+        tracing::debug!("Candidate for removal: {}", candidate.spec().uuid);
+
+        for repl in &self.healthy {
+            let Some(node) = repl.node_spec() else {
+                continue;
+            };
+            if restricted_nodes.get(node.id()).is_none()
+                && (repl.valid_topology() || !candidate.valid_topology())
+            {
+                replicas_in_unrestricted_nodes = true;
+                break;
+            }
+        }
+
+        if replicas_in_unrestricted_nodes {
+            Some(Some(candidate))
+        } else {
+            Some(None)
+        }
+    }
 
     fn new(context: GetChildForRemovalContext, items: Vec<ReplicaItem>) -> Self {
         let has_info = context.nexus_info.is_some();