DAOS-18592 control: Only use joined ranks when calculating pool size (#17580)

In order to avoid failing pool create with storage percentage (-z X%)
when ranks have been stopped, only take into account joined ranks when
calculating maximum available pool sizes.

Signed-off-by: Tom Nabarro <thomas.nabarro@hpe.com>

Author: tanabarr

src/control/lib/control/pool.go

@@ -1188,16 +1188,8 @@ func ListPools(ctx context.Context, rpcClient UnaryInvoker, req *ListPoolsReq) (
 
 type rankFreeSpaceMap map[ranklist.Rank]uint64
 
-type filterRankFn func(rank ranklist.Rank) bool
-
-func newFilterRankFunc(ranks ranklist.RankList) filterRankFn {
-	return func(rank ranklist.Rank) bool {
-		return len(ranks) == 0 || rank.InList(ranks)
-	}
-}
-
 // Add namespace ranks to rankNVMeFreeSpace map and return minimum free available SCM namespace bytes.
-func processSCMSpaceStats(log debugLogger, filterRank filterRankFn, scmNamespaces storage.ScmNamespaces, rankNVMeFreeSpace rankFreeSpaceMap) (uint64, error) {
+func processSCMSpaceStats(log debugLogger, ranks ranklist.RankList, scmNamespaces storage.ScmNamespaces, rankNVMeFreeSpace rankFreeSpaceMap) (uint64, error) {
 	scmBytes := uint64(math.MaxUint64)
 
 	// Realistically there should only be one-per-rank but handle the case for multiple anyway.
@@ -1207,7 +1199,7 @@ func processSCMSpaceStats(log debugLogger, filterRank filterRankFn, scmNamespace
 				scmNamespace.UUID, scmNamespace.BlockDevice, scmNamespace.Name)
 		}
 
-		if !filterRank(scmNamespace.Mount.Rank) {
+		if !scmNamespace.Mount.Rank.InList(ranks) {
 			log.Debugf("Skipping SCM device %s (bdev %s, name %s, rank %d) not in ranklist",
 				scmNamespace.UUID, scmNamespace.BlockDevice, scmNamespace.Name,
 				scmNamespace.Mount.Rank)
@@ -1233,7 +1225,7 @@ func processSCMSpaceStats(log debugLogger, filterRank filterRankFn, scmNamespace
 }
 
 // Add NVMe free bytes to rankNVMeFreeSpace map.
-func processNVMeSpaceStats(log debugLogger, filterRank filterRankFn, nvmeControllers storage.NvmeControllers, rankNVMeFreeSpace rankFreeSpaceMap) error {
+func processNVMeSpaceStats(log debugLogger, ranks ranklist.RankList, nvmeControllers storage.NvmeControllers, rankNVMeFreeSpace rankFreeSpaceMap) error {
 	for _, controller := range nvmeControllers {
 		for _, smdDevice := range controller.SmdDevices {
 			msgDev := fmt.Sprintf("SMD device %s (rank %d, ctrlr %s", smdDevice.UUID,
@@ -1258,7 +1250,7 @@ func processNVMeSpaceStats(log debugLogger, filterRank filterRankFn, nvmeControl
 					controller.NvmeState.String())
 			}
 
-			if !filterRank(smdDevice.Rank) {
+			if !smdDevice.Rank.InList(ranks) {
 				log.Debugf("Skipping %s, not in ranklist", msgDev)
 				continue
 			}
@@ -1288,10 +1280,34 @@ func getMaxPoolSize(ctx context.Context, rpcClient UnaryInvoker, createReq *Pool
 		return 0, 0, errors.New("invalid mem-ratio, should not be greater than one")
 	}
 
-	// Verify that the DAOS system is ready before attempting to query storage.
-	if _, err := SystemQuery(ctx, rpcClient, &SystemQueryReq{}); err != nil {
-		return 0, 0, err
+	// Verify that the DAOS system is ready before attempting to query storage and record joined.
+	queryResp, err := SystemQuery(ctx, rpcClient, &SystemQueryReq{})
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "getMaxPoolSize: SystemQuery")
+	}
+	joinedRanks := ranklist.RankList{}
+	for _, member := range queryResp.Members {
+		if member.State == system.MemberStateJoined {
+			joinedRanks = append(joinedRanks, member.Rank)
+		}
+	}
+
+	// Refuse if any requested ranks are not joined, update ranklist to contain only joined ranks.
+	filterRanks := ranklist.RankList{}
+	if len(createReq.Ranks) == 0 {
+		filterRanks = joinedRanks
+	} else {
+		for _, rank := range createReq.Ranks {
+			if !rank.InList(joinedRanks) {
+				return 0, 0, errors.Errorf("specified rank %d is not joined", rank)
+			}
+			filterRanks = append(filterRanks, rank)
+		}
 	}
+	slices.Sort(filterRanks)
+	rpcClient.Debugf("requested/joined/filter ranks: %v/%v/%v", createReq.Ranks, joinedRanks,
+		filterRanks)
+	createReq.Ranks = filterRanks
 
 	scanReq := &StorageScanReq{
 		Usage:    true,
@@ -1307,8 +1323,6 @@ func getMaxPoolSize(ctx context.Context, rpcClient UnaryInvoker, createReq *Pool
 		return 0, 0, errors.New("Empty host storage response from StorageScan")
 	}
 
-	// Generate function to verify a rank is in the provided rank slice.
-	filterRank := newFilterRankFunc(ranklist.RankList(createReq.Ranks))
 	rankNVMeFreeSpace := make(rankFreeSpaceMap)
 	scmBytes := uint64(math.MaxUint64)
 	for _, key := range scanResp.HostStorage.Keys() {
@@ -1319,7 +1333,7 @@ func getMaxPoolSize(ctx context.Context, rpcClient UnaryInvoker, createReq *Pool
 				scanResp.HostStorage[key].HostSet.String())
 		}
 
-		sb, err := processSCMSpaceStats(rpcClient, filterRank, hostStorage.ScmNamespaces, rankNVMeFreeSpace)
+		sb, err := processSCMSpaceStats(rpcClient, filterRanks, hostStorage.ScmNamespaces, rankNVMeFreeSpace)
 		if err != nil {
 			return 0, 0, err
 		}
@@ -1328,7 +1342,7 @@ func getMaxPoolSize(ctx context.Context, rpcClient UnaryInvoker, createReq *Pool
 			scmBytes = sb
 		}
 
-		if err := processNVMeSpaceStats(rpcClient, filterRank, hostStorage.NvmeDevices, rankNVMeFreeSpace); err != nil {
+		if err := processNVMeSpaceStats(rpcClient, filterRanks, hostStorage.NvmeDevices, rankNVMeFreeSpace); err != nil {
 			return 0, 0, err
 		}
 	}