support recovery when server down

eraft-io · Jul 9, 2023 · e867c5e · e867c5e
1 parent 3090f56
commit e867c5e
Show file tree

Hide file tree

Showing 3 changed files with 79 additions and 57 deletions.
diff --git a/src/client.cc b/src/client.cc
@@ -50,7 +50,7 @@ Client::Client(std::string meta_addrs)
     auto chan_ =
         grpc::CreateChannel(meta_node_addr, grpc::InsecureChannelCredentials());
     std::unique_ptr<ERaftKv::Stub> stub_(ERaftKv::NewStub(chan_));
-    this->stubs_[meta_node_addr] = std::move(stub_);
+    this->meta_stubs_[meta_node_addr] = std::move(stub_);
   }
   // sync config
   SyncClusterConfig();
@@ -74,42 +74,50 @@ std::string Client::GetLeaderAddr(std::string partion_key) {
       if (key_slot == sl.id()) {
         // find sg leader addr
         for (auto server : sg.servers()) {
-          if (server.id() == sg.leader_id()) {
-            ClientContext                   context;
-            eraftkv::ClusterConfigChangeReq req;
-            req.set_change_type(eraftkv::ChangeType::MetaMembersQuery);
-            eraftkv::ClusterConfigChangeResp resp;
-            auto status = stubs_[server.address()]->ClusterConfigChange(
-                &context, req, &resp);
-            for (int i = 0; i < resp.shard_group(0).servers_size(); i++) {
-              if (resp.shard_group(0).leader_id() ==
-                  resp.shard_group(0).servers(i).id()) {
-                leader_address = resp.shard_group(0).servers(i).address();
-              }
+          ClientContext                         context;
+          std::chrono::system_clock::time_point deadline =
+              std::chrono::system_clock::now() + std::chrono::seconds(1);
+          context.set_deadline(deadline);
+          eraftkv::ClusterConfigChangeReq req;
+          req.set_change_type(eraftkv::ChangeType::MetaMembersQuery);
+          eraftkv::ClusterConfigChangeResp resp;
+          auto status = stubs_[server.address()]->ClusterConfigChange(
+              &context, req, &resp);
+          if (!status.ok()) {
+            continue;
+          }
+          for (int i = 0; i < resp.shard_group(0).servers_size(); i++) {
+            if (resp.shard_group(0).leader_id() ==
+                resp.shard_group(0).servers(i).id()) {
+              leader_address = resp.shard_group(0).servers(i).address();
             }
           }
         }
       }
     }
   }
-
   return leader_address;
 }
 
 EStatus Client::SyncClusterConfig() {
-  ClientContext                   context;
-  eraftkv::ClusterConfigChangeReq req;
-  req.set_change_type(eraftkv::ChangeType::ShardsQuery);
-  auto status_ = this->stubs_.begin()->second->ClusterConfigChange(
-      &context, req, &cluster_conf_);
-  for (auto sg : cluster_conf_.shard_group()) {
-    for (auto server : sg.servers()) {
-      auto chan_ = grpc::CreateChannel(server.address(),
-                                       grpc::InsecureChannelCredentials());
-      std::unique_ptr<ERaftKv::Stub> stub_(ERaftKv::NewStub(chan_));
-      this->stubs_[server.address()] = std::move(stub_);
+  for (auto it = this->meta_stubs_.begin(); it != this->meta_stubs_.end();
+       it++) {
+    ClientContext                   context;
+    eraftkv::ClusterConfigChangeReq req;
+    req.set_change_type(eraftkv::ChangeType::ShardsQuery);
+    auto status_ =
+        it->second->ClusterConfigChange(&context, req, &cluster_conf_);
+    if (!status_.ok()) {
+      continue;
+    }
+    for (auto sg : cluster_conf_.shard_group()) {
+      for (auto server : sg.servers()) {
+        auto chan_ = grpc::CreateChannel(server.address(),
+                                         grpc::InsecureChannelCredentials());
+        std::unique_ptr<ERaftKv::Stub> stub_(ERaftKv::NewStub(chan_));
+        this->stubs_[server.address()] = std::move(stub_);
+      }
     }
   }
-
-  return status_.ok() ? EStatus::kOk : EStatus::kError;
+  return EStatus::kOk;
 }
diff --git a/src/client.h b/src/client.h
@@ -42,6 +42,8 @@ class Client : public StreamSocket {
 
   std::map<std::string, std::unique_ptr<ERaftKv::Stub> > stubs_;
 
+  std::map<std::string, std::unique_ptr<ERaftKv::Stub> > meta_stubs_;
+
   std::string leader_addr_;
 
   eraftkv::ClusterConfigChangeResp cluster_conf_;

diff --git a/src/info_command_handler.cc b/src/info_command_handler.cc
@@ -14,37 +14,49 @@
 
 EStatus InfoCommandHandler::Execute(const std::vector<std::string>& params,
                                     Client*                         cli) {
-  ClientContext                   context;
-  eraftkv::ClusterConfigChangeReq req;
-  req.set_change_type(eraftkv::ChangeType::MetaMembersQuery);
-  eraftkv::ClusterConfigChangeResp resp;
-  auto status = cli->stubs_[StringUtil::Split(cli->meta_addrs_, ',')[0]]
-                    ->ClusterConfigChange(&context, req, &resp);
-  std::string info_str;
-  for (int i = 0; i < resp.shard_group(0).servers_size(); i++) {
-    info_str += "meta server: \r\n";
-    info_str += "server_id: ";
-    info_str += std::to_string(resp.shard_group(0).servers(i).id());
-    info_str += ",server_address: ";
-    info_str += resp.shard_group(0).servers(i).address();
-    resp.shard_group(0).servers(i).server_status() == eraftkv::ServerStatus::Up
-        ? info_str += ",status: Running"
-        : info_str += ",status: Down";
-    resp.shard_group(0).leader_id() == resp.shard_group(0).servers(i).id()
-        ? info_str += ",Role: Leader"
-        : info_str += ",Role: Follower";
-    info_str += "\r\n";
-  }
-  std::string reply_buf;
-  reply_buf += "$";
-  reply_buf += std::to_string(info_str.size());
-  reply_buf += "\r\n";
-  reply_buf += info_str;
-  reply_buf += "\r\n";
-  cli->reply_.PushData(reply_buf.c_str(), reply_buf.size());
-  cli->SendPacket(cli->reply_);
+  for (auto it = cli->meta_stubs_.begin(); it != cli->meta_stubs_.end(); it++) {
+    ClientContext                         context;
+    std::chrono::system_clock::time_point deadline =
+        std::chrono::system_clock::now() + std::chrono::seconds(1);
+    context.set_deadline(deadline);
+    eraftkv::ClusterConfigChangeReq req;
+    req.set_change_type(eraftkv::ChangeType::MetaMembersQuery);
+    eraftkv::ClusterConfigChangeResp resp;
+    auto status = it->second->ClusterConfigChange(&context, req, &resp);
+    if (!status.ok()) {
+      continue;
+    }
+    std::string info_str;
+    for (int i = 0; i < resp.shard_group(0).servers_size(); i++) {
+      info_str += "meta server: \r\n";
+      info_str += "server_id: ";
+      info_str += std::to_string(resp.shard_group(0).servers(i).id());
+      info_str += ",server_address: ";
+      info_str += resp.shard_group(0).servers(i).address();
+      resp.shard_group(0).servers(i).server_status() ==
+              eraftkv::ServerStatus::Up
+          ? info_str += ",status: Running"
+          : info_str += ",status: Down";
+      resp.shard_group(0).leader_id() == resp.shard_group(0).servers(i).id()
+          ? info_str += ",Role: Leader"
+          : info_str += ",Role: Follower";
+      info_str += "\r\n";
+    }
+    std::string reply_buf;
+    reply_buf += "$";
+    reply_buf += std::to_string(info_str.size());
+    reply_buf += "\r\n";
+    reply_buf += info_str;
+    reply_buf += "\r\n";
+
+    cli->reply_.PushData(reply_buf.c_str(), reply_buf.size());
+    cli->SendPacket(cli->reply_);
 
-  cli->_Reset();
+    cli->_Reset();
+    if (status.ok()) {
+      break;
+    }
+  }
   return EStatus::kOk;
 }