schedule pendding online chunkserver

src/mds/schedule/copySetScheduler.cpp

@@ -32,33 +32,97 @@ namespace curve {
 namespace mds {
 namespace schedule {
 int CopySetScheduler::Schedule() {
-    LOG(INFO) << "copysetScheduler begin";
+    LOG(INFO) << "schedule: copysetScheduler begin";
 
-    int res = 0;
+    int oneRoundGenOp = 0;
     for (auto lid : topo_->GetLogicalpools()) {
-        res = DoCopySetSchedule(lid);
+        oneRoundGenOp += DoCopySetSchedule(lid);
     }
-    return res;
+
+    LOG(INFO) << "schedule: copysetScheduler end, generate operator num "
+              << oneRoundGenOp;
+    return oneRoundGenOp;
 }
 
-int CopySetScheduler::DoCopySetSchedule(PoolIdType lid) {
-    // 1. collect the chunkserver list and copyset list of the cluster, then
-    //    collect copyset on every online chunkserver
-    auto copysetList = topo_->GetCopySetInfosInLogicalPool(lid);
-    auto chunkserverList = topo_->GetChunkServersInLogicalPool(lid);
-    std::map<ChunkServerIdType, std::vector<CopySetInfo>> distribute;
-    SchedulerHelper::CopySetDistributionInOnlineChunkServer(
-        copysetList, chunkserverList, &distribute);
-    if (distribute.empty()) {
-        LOG(WARNING) << "no not-retired chunkserver in topology";
-        return UNINTIALIZE_ID;
+int CopySetScheduler::PenddingCopySetSchedule(const std::map<ChunkServerIdType,
+                                    std::vector<CopySetInfo>> &distribute) {
+    int oneRoundGenOp = 0;
+    // for every chunkserver, find one copyset to migrate out
+    for (auto it = distribute.begin(); it != distribute.end(); it++) {
+        ChunkServerIdType source = it->first;
+        int copysetNum = it->second.size();
+        if (copysetNum == 0) {
+            continue;
+        }
+
+        // find one copyset to migrate out from source chunkserver
+        for (auto info : it->second) {
+            // does not meet the basic conditions
+            if (!CopySetSatisfiyBasicMigrationCond(info)) {
+                continue;
+            }
+
+            auto target = SelectBestPlacementChunkServer(info, source);
+            if (target == UNINTIALIZE_ID) {
+                LOG(WARNING) << "copysetScheduler can not select chunkServer "
+                                "to migrate " << info.CopySetInfoStr()
+                            << ", which replica: " << source << " is pendding";
+                continue;
+            }
+
+            Operator op = operatorFactory.CreateChangePeerOperator(
+                info, source, target, OperatorPriority::HighPriority);
+            op.timeLimit = std::chrono::seconds(changeTimeSec_);
+
+            if (AddOperatorAndCreateCopyset(op, info, target)) {
+                oneRoundGenOp++;
+            }
+        }
+    }
+
+    if (oneRoundGenOp != 0) {
+        LOG(INFO) << "pendding copyset scheduler migrate " << oneRoundGenOp
+                        << " copyset at this round";
+    }
+
+    return oneRoundGenOp;
+}
+
+bool CopySetScheduler::AddOperatorAndCreateCopyset(const Operator &op,
+                                            const CopySetInfo &choose,
+                                            const ChunkServerIdType &target) {
+    // add operator
+    if (!opController_->AddOperator(op)) {
+        LOG(INFO) << "copysetSchduler add op " << op.OpToString()
+                    << " fail, copyset has already has operator"
+                    << " or operator num exceeds the limit.";
+        return false;
     }
 
+    // create copyset
+    if (!topo_->CreateCopySetAtChunkServer(choose.id, target)) {
+        LOG(ERROR) << "copysetScheduler create " << choose.CopySetInfoStr()
+                    << " on chunkServer: " << target
+                    << " error, delete operator" << op.OpToString();
+        opController_->RemoveOperator(choose.id);
+        return false;
+    }
+
+    LOG(INFO) << "copysetScheduler create " << choose.CopySetInfoStr()
+                << "on chunkserver:" << target
+                << " success. generator op: "
+                << op.OpToString() << "success";
+    return true;
+}
+
+int CopySetScheduler::NormalCopySetSchedule(const std::map<ChunkServerIdType,
+                                    std::vector<CopySetInfo>> &distribute) {
     // 2. measure the average, range and standard deviation of number of copyset
     //    on chunkservers
     float avg;
     int range;
     float stdvariance;
+    int oneRoundGenOp = 0;
     StatsCopysetDistribute(distribute, &avg, &range, &stdvariance);
     /**
      * 3. Set migration condition
@@ -83,37 +147,53 @@ int CopySetScheduler::DoCopySetSchedule(PoolIdType lid) {
      **/
     ChunkServerIdType source = UNINTIALIZE_ID;
     if (range <= avg * copysetNumRangePercent_) {
-        return source;
+        return oneRoundGenOp;
     }
 
     Operator op;
     ChunkServerIdType target = UNINTIALIZE_ID;
     CopySetInfo choose;
     // this function call will select the source, target and the copyset
     if (CopySetMigration(distribute, &op, &source, &target, &choose)) {
-        // add operator
-        if (!opController_->AddOperator(op)) {
-            LOG(INFO) << "copysetSchduler add op " << op.OpToString()
-                      << " fail, copyset has already has operator";
+        if (AddOperatorAndCreateCopyset(op, choose, target)) {
+            oneRoundGenOp++;
         }
+    }
+
+    return oneRoundGenOp;
+}
 
-        // create copyset
-        if (!topo_->CreateCopySetAtChunkServer(choose.id, target)) {
-            LOG(ERROR) << "copysetScheduler create " << choose.CopySetInfoStr()
-                       << " on chunkServer: " << target
-                       << " error, delete operator" << op.OpToString();
-            opController_->RemoveOperator(choose.id);
-        } else {
-            LOG(INFO) << "copysetScheduler create " << choose.CopySetInfoStr()
-                      << "on chunkserver:" << target
-                      << " success. generator op: "
-                      << op.OpToString() << "success";
+int CopySetScheduler::DoCopySetSchedule(PoolIdType lid) {
+    // 1. collect the chunkserver list and copyset list of the cluster, then
+    //    collect copyset on every online chunkserver
+    auto copysetList = topo_->GetCopySetInfosInLogicalPool(lid);
+    auto chunkserverList = topo_->GetChunkServersInLogicalPool(lid);
+
+    std::map<ChunkServerIdType, std::vector<CopySetInfo>> penddingDistribute;
+    SchedulerHelper::GetCopySetDistributionInOnlineChunkServer(
+        copysetList, chunkserverList, &penddingDistribute);
+    SchedulerHelper::FilterCopySetDistributions(ChunkServerStatus::PENDDING,
+                    chunkserverList, &penddingDistribute);
+    if (!penddingDistribute.empty()) {
+        int oneRoundGenOp = PenddingCopySetSchedule(penddingDistribute);
+        // If generate pendding copy set schedule, return here.
+        if (oneRoundGenOp != 0) {
+            return oneRoundGenOp;
         }
     }
 
-    LOG_EVERY_N(INFO, 20) << "copysetScheduler is continually adjusting";
-    LOG(INFO) << "copysetScheduler end.";
-    return static_cast<int>(source);
+    // If no pendding copyset schedule operator generated,
+    // run NormalCopySetSchedule
+    std::map<ChunkServerIdType, std::vector<CopySetInfo>> normalDistribute;
+    SchedulerHelper::GetCopySetDistributionInOnlineChunkServer(
+        copysetList, chunkserverList, &normalDistribute);
+    SchedulerHelper::FilterCopySetDistributions(ChunkServerStatus::READWRITE,
+                    chunkserverList, &normalDistribute);
+    if (normalDistribute.empty()) {
+        LOG(WARNING) << "no not-retired chunkserver in topology";
+        return 0;
+    }
+    return NormalCopySetSchedule(normalDistribute);
 }
 
 void CopySetScheduler::StatsCopysetDistribute(

src/mds/schedule/leaderScheduler.cpp

@@ -34,12 +34,15 @@ namespace curve {
 namespace mds {
 namespace schedule {
 int LeaderScheduler::Schedule() {
-    LOG(INFO) << "leaderScheduler begin.";
-
+    LOG(INFO) << "schedule: leaderScheduler begin.";
+    int oneRoundGenOp = 0;
     for (auto lid : topo_->GetLogicalpools()) {
-        DoLeaderSchedule(lid);
+        oneRoundGenOp += DoLeaderSchedule(lid);
     }
-    return 1;
+
+    LOG(INFO) << "schedule: leaderScheduler end, generate operator num "
+              << oneRoundGenOp;
+    return oneRoundGenOp;
 }
 
 int LeaderScheduler::DoLeaderSchedule(PoolIdType lid) {
@@ -57,7 +60,8 @@ int LeaderScheduler::DoLeaderSchedule(PoolIdType lid) {
     std::shuffle(csInfos.begin(), csInfos.end(), g);
 
     for (auto csInfo : csInfos) {
-        if (csInfo.IsOffline()) {
+        // skip offline chunkserver or pendding chunkserver
+        if (csInfo.IsOffline() || csInfo.IsPendding()) {
             continue;
         }
 
@@ -177,6 +181,12 @@ bool LeaderScheduler::transferLeaderOut(ChunkServerIdType source, int count,
                 break;
             }
 
+            // can not transfer to pendding chunkserver
+            if (csInfo.IsPendding()) {
+                continue;
+            }
+
+            // can not transfer to myself
             if (source == peerInfo.id) {
                 continue;
             }

src/mds/schedule/rapidLeaderScheduler.cpp

@@ -120,9 +120,10 @@ bool RapidLeaderScheduler::LeaderStatInSpecifiedLogicalPool(
     }
 
     // get copyset info list for every chunkserver
-    SchedulerHelper::CopySetDistributionInOnlineChunkServer(
+    SchedulerHelper::GetCopySetDistributionInOnlineChunkServer(
         copysetVec, chunkserverVec, &stat->distribute);
-
+    SchedulerHelper::FilterCopySetDistributions(ChunkServerStatus::READWRITE,
+                    chunkserverVec, &stat->distribute);
     // calculate average leader number for every chunkserver
     stat->avgLeaderNum = copysetVec.size() / chunkserverVec.size();
 

src/mds/schedule/scheduler.cpp

@@ -256,7 +256,8 @@ ChunkServerIdType Scheduler::SelectRedundantReplicaToRemove(
     // the one to remove should be selected from BCDE
     // 2.2 greater
     // any chunkserver to remove will satisfy the requirement, thus we first
-    // consider the one not online, then consider the scatter-width
+    // consider the one not online, then the one is pendding ,
+    // then consider the scatter-width
     // for example, if the replicas are:
     // A(zone1) B(zone2) C(zone3) D(zone4) E(zone4)
     // we can remove anyone of it
@@ -273,7 +274,7 @@ ChunkServerIdType Scheduler::SelectRedundantReplicaToRemove(
         }
     }
 
-    // consider offline chunkservers first
+    // consider offline or pendding chunkservers first
     for (auto cs : candidateChunkServer) {
         ChunkServerInfo csInfo;
         if (!topo_->GetChunkServerInfo(cs, &csInfo)) {
@@ -283,15 +284,22 @@ ChunkServerIdType Scheduler::SelectRedundantReplicaToRemove(
             return UNINTIALIZE_ID;
         }
 
-        // chunkserver is not offline
+        // chunkserver is offline
         if (csInfo.IsOffline()) {
             LOG(WARNING) << "scheduler choose to remove offline chunkServer "
                          << cs << " from " << copySetInfo.CopySetInfoStr();
             return cs;
         }
+
+        // chunkserver is pendding
+        if (csInfo.IsPendding()) {
+            LOG(WARNING) << "scheduler choose to remove pendding chunkServer "
+                         << cs << " from " << copySetInfo.CopySetInfoStr();
+            return cs;
+        }
     }
 
-    // if all chunkserver are online, select according to the impact on scatter-width //NOLINT
+    // if all chunkserver are online and not pendding, select according to the impact on scatter-width //NOLINT
     // first rank chunkser by the number of copyset on it
     std::map<ChunkServerIdType, std::vector<CopySetInfo>> distribute;
     std::vector<std::pair<ChunkServerIdType, std::vector<CopySetInfo>>> desc;

src/mds/schedule/scheduler.h

@@ -171,7 +171,7 @@ class CopySetScheduler : public Scheduler {
      * @brief Schedule Generating operator according to
      *        the condition of the cluster
      *
-     * @return chunkserver to add, a value for POC
+     * @return operator num generated
      */
     int Schedule() override;
 
@@ -245,6 +245,36 @@ class CopySetScheduler : public Scheduler {
      */
     bool CopySetSatisfiyBasicMigrationCond(const CopySetInfo &info);
 
+    /**
+     * @brief migrate one copyset between online & no pendding chunkservers
+     *
+     * @param[in] distribute Copyset on every online && no pendding chunkserver
+     *
+     * @return Source node of the migration
+     */
+    int NormalCopySetSchedule(const std::map<ChunkServerIdType,
+                                    std::vector<CopySetInfo>> &distribute);
+
+    /**
+     * @brief migrate one copyset from online && pendding chunkserver to
+     *        other healty chunkserver
+     * @param[in] distribute Copyset on every online && pendding chunkserver
+     * @return migrate copyset num
+     */
+    int PenddingCopySetSchedule(const std::map<ChunkServerIdType,
+                                    std::vector<CopySetInfo>> &distribute);
+
+    /**
+     * @brief add op to opController and create copyset at chunkserver
+     * @param[in] op the oprator to add to opController
+     * @param[in] choose the copyset need to schedule
+     * @param[in] target the chunkserver want to add to this copyset
+     * @return return true if sucess, retrun false if not
+     */
+    bool AddOperatorAndCreateCopyset(const Operator &op,
+                                     const CopySetInfo &choose,
+                                     const ChunkServerIdType &target);
+
  private:
     // Running interval of CopySetScheduler
     int64_t runInterval_;

src/mds/schedule/scheduler_helper.cpp

@@ -353,7 +353,7 @@ bool SchedulerHelper::InvovledReplicasSatisfyScatterWidthAfterMigration(
     return allSatisfy;
 }
 
-void SchedulerHelper::CopySetDistributionInOnlineChunkServer(
+void SchedulerHelper::GetCopySetDistributionInOnlineChunkServer(
         const std::vector<CopySetInfo> &copysetList,
         const std::vector<ChunkServerInfo> &chunkserverList,
         std::map<ChunkServerIdType, std::vector<CopySetInfo>> *out) {
@@ -370,15 +370,30 @@ void SchedulerHelper::CopySetDistributionInOnlineChunkServer(
 
     // remove offline chunkserver, and report empty list for empty chunkserver
     for (auto item : chunkserverList) {
+        // remove offline chunkserver
         if (item.IsOffline()) {
             out->erase(item.info.id);
             continue;
         }
+
+        // report empty list for chunkserver with no copyset
         if (out->find(item.info.id) == out->end()) {
             (*out)[item.info.id] = std::vector<CopySetInfo>{};
         }
     }
 }
+
+void SchedulerHelper::FilterCopySetDistributions(const ChunkServerStatus status,
+        const std::vector<ChunkServerInfo> &chunkserverList,
+        std::map<ChunkServerIdType, std::vector<CopySetInfo>> *distributions) {
+    for (auto item : chunkserverList) {
+        if (item.status != status) {
+            distributions->erase(item.info.id);
+        }
+    }
+    return;
+}
+
 }  // namespace schedule
 }  // namespace mds
 }  // namespace curve