优化并去掉二次寻址 (#34)

* feat: c++ 对齐其他语言SDK

1. 限流接口支持传入method
2. 去掉grpc以及动态权重调整的代码
3. 支持非二次寻址
4. 支持环境变量配置

* feat：修复用例失败问题

* feat: 默认地址导致用例失败问题

* feat: 优化并去掉二次寻址

Author: andrewshan

polaris/context/context.cpp

@@ -45,8 +45,9 @@ Context* Context::Create(Config* config, ContextMode mode) {
   if (context_impl->Init(config, context.get(), mode) != kReturnOk) {
     return nullptr;
   }
+  const PolarisCluster& discover_cluster = context_impl->GetDiscoverService();
   // Polaris discover先请求一下
-  if (context_impl->InitSystemService(context_impl->GetDiscoverService()) != kReturnOk) {
+  if (!discover_cluster.service_.name_.empty() && context_impl->InitSystemService(discover_cluster) != kReturnOk) {
     return nullptr;
   }
   // 如果有设置Metric Cluster则提前获取

polaris/context/context_impl.cpp

@@ -230,6 +230,8 @@ ReturnCode ContextImpl::InitSystemConfig(Config* system_config) {
 
 ReturnCode ContextImpl::InitSystemService(const PolarisCluster& cluster) {
   ServiceContext* service_context = this->GetServiceContext(cluster.service_);
+  POLARIS_LOG(LOG_ERROR, "init system service for service[%s/%s] failed", cluster.service_.namespace_.c_str(),
+                cluster.service_.name_.c_str());
   if (service_context == nullptr) {
     POLARIS_LOG(LOG_ERROR, "create service context for service[%s/%s] failed", cluster.service_.namespace_.c_str(),
                 cluster.service_.name_.c_str());

polaris/monitor/monitor_reporter.cpp

@@ -831,6 +831,9 @@ bool MonitorReporter::GetInstance(ReportBase* report_data) {
   POLARIS_ASSERT(report_data != nullptr);
   POLARIS_ASSERT(report_data->instance_ == nullptr);
   const PolarisCluster& monitor_cluster = context_->GetContextImpl()->GetMonitorService();
+  if (monitor_cluster.service_.name_.empty()) {
+    return false;
+  }
   Criteria criteria;
   criteria.ignore_half_open_ = (rand() % 10 != 0);  // 10分之一的概率选择半开节点
   ReturnCode ret_code = ConsumerApiImpl::GetSystemServer(context_, monitor_cluster.service_, criteria,

polaris/plugin/server_connector/grpc_server_connector.cpp

@@ -335,7 +335,8 @@ bool GrpcServerConnector::SendDiscoverRequest(ServiceListener& service_listener)
   }
   if (discover_stream_state_ != kDiscoverStreamInit) {
     const ServiceKey& discover_service = context_->GetContextImpl()->GetDiscoverService().service_;
-    if (isEmpty(discover_service)) {
+    if (discover_service.name_.empty()) {
+      POLARIS_LOG(LOG_INFO, "discover service is empty, state transive to DiscoverStreamInit");
       discover_stream_state_ = kDiscoverStreamInit;
     } else {
       if (service_key.name_ != discover_service.name_ || service_key.namespace_ != discover_service.namespace_) {
@@ -572,6 +573,10 @@ ReturnCode GrpcServerConnector::SelectInstance(const ServiceKey& service_key, ui
   return ConsumerApiImpl::GetSystemServer(context_, service_key, criteria, *instance, timeout);
 }
 
+SeedServer& GrpcServerConnector::SelectSeed() {
+  return server_lists_[rand() % server_lists_.size()];
+}
+
 void GrpcServerConnector::ServerSwitch() {
   if (server_switch_state_ == kServerSwitchNormal       // 服务调用出错或超时触发切换
       || server_switch_state_ == kServerSwitchBegin) {  // 切换后异步连接回调触发重新
@@ -590,12 +595,12 @@ void GrpcServerConnector::ServerSwitch() {
   // 选择一个服务器
   std::string host;
   int port = 0;
-  if (discover_stream_state_ >= kDiscoverStreamGetInstance) {  // 说明内部服务已经返回
+  const ServiceKey& discover_service = context_->GetContextImpl()->GetDiscoverService().service_;
+  if (!discover_service.name_.empty() && discover_stream_state_ >= kDiscoverStreamGetInstance) {  // 说明内部服务已经返回
     if (discover_instance_ != nullptr) {
       delete discover_instance_;
       discover_instance_ = nullptr;
     }
-    const ServiceKey& discover_service = context_->GetContextImpl()->GetDiscoverService().service_;
     bool ignore_half_open = server_switch_state_ != kServerSwitchPeriodic;  // 周期切换才选半开节点
     ReturnCode ret_code = SelectInstance(discover_service, 0, &discover_instance_, ignore_half_open);
     if (ret_code == kReturnOk) {
@@ -612,7 +617,7 @@ void GrpcServerConnector::ServerSwitch() {
   }
   if (host.empty()) {
     discover_stream_state_ = kDiscoverStreamNotInit;
-    SeedServer& server = server_lists_[rand() % server_lists_.size()];
+    SeedServer& server = SelectSeed();
     host = server.ip_;
     port = server.port_;
     POLARIS_LOG(LOG_INFO, "discover stream switch to seed server[%s:%d]", host.c_str(), port);
@@ -837,11 +842,19 @@ bool GrpcServerConnector::GetInstance(BlockRequest* block_request) {
   POLARIS_ASSERT(block_request != nullptr);
   POLARIS_ASSERT(block_request->instance_ == nullptr);
   const ServiceKey& service = GetPolarisService(context_, block_request->request_type_);
+  if (service.name_.empty()) {
+    SeedServer& seedServer = SelectSeed();
+    block_request->host_ = seedServer.ip_;
+    block_request->port_ = seedServer.port_;
+    return true;
+  }
   ReturnCode ret_code = SelectInstance(service, block_request->request_timeout_, &block_request->instance_);
   if (ret_code == kReturnOk) {
     POLARIS_ASSERT(block_request->instance_ != nullptr);
     POLARIS_LOG(LOG_DEBUG, "get server:%s:%d for %s", block_request->instance_->GetHost().c_str(),
                 block_request->instance_->GetPort(), PolarisRequestTypeStr(block_request->request_type_));
+    block_request->host_ = block_request->instance_->GetHost();
+    block_request->port_ = block_request->instance_->GetPort();
     return true;
   } else {
     POLARIS_ASSERT(block_request->instance_ == nullptr);
@@ -852,7 +865,9 @@ bool GrpcServerConnector::GetInstance(BlockRequest* block_request) {
 }
 
 void GrpcServerConnector::UpdateCallResult(BlockRequest* block_request) {
-  POLARIS_ASSERT(block_request->instance_ != nullptr);
+  if(block_request->instance_ == nullptr) {
+    return;
+  }
   const ServiceKey& service = GetPolarisService(context_, block_request->request_type_);
   CallRetStatus status = kCallRetOk;
   if (kServerCodeConnectError <= block_request->server_code_ &&
@@ -876,7 +891,9 @@ BlockRequest::BlockRequest(PolarisRequestType request_type, GrpcServerConnector&
       message_(nullptr),
       promise_(nullptr),
       instance_(nullptr),
-      grpc_client_(nullptr) {}
+      host_(""),
+      port_(0),
+      grpc_client_(nullptr)  {}
 
 BlockRequest::~BlockRequest() {
   if (instance_ != nullptr) {
@@ -934,18 +951,18 @@ bool BlockRequest::PrepareClient() {
 
   // 建立grpc客户端，并尝试连接
   grpc_client_ = new grpc::GrpcClient(connector_.GetReactor());
-  if (!grpc_client_->ConnectTo(instance_->GetHost(), instance_->GetPort()) ||
+  if (!grpc_client_->ConnectTo(host_, port_) ||
       !grpc_client_->WaitConnected(request_timeout_)) {
     POLARIS_LOG(LOG_ERROR, "%s connect to server[%s:%d] timeout", PolarisRequestTypeStr(request_type_),
-                instance_->GetHost().c_str(), instance_->GetPort());
+                host_.c_str(), port_);
     server_code_ = kServerCodeConnectError;
     connector_.UpdateCallResult(this);
     return false;
   }
   uint64_t use_time = Time::GetCoarseSteadyTimeMs() - begin_time;
   if (use_time >= request_timeout_) {
     POLARIS_LOG(LOG_ERROR, "%s connect to server[%s:%d] timeout", PolarisRequestTypeStr(request_type_),
-                instance_->GetHost().c_str(), instance_->GetPort());
+                host_.c_str(), port_);
     server_code_ = kServerCodeConnectError;
     connector_.UpdateCallResult(this);
     return false;
@@ -1020,6 +1037,8 @@ AsyncRequest::AsyncRequest(Reactor& reactor, GrpcServerConnector* connector, Pol
       timeout_(timeout),
       callback_(callback),
       server_(nullptr),
+      host_(""),
+      port_(0),
       client_(nullptr),
       timing_task_(connector->GetReactor().TimingTaskEnd()) {}
 
@@ -1046,16 +1065,23 @@ bool AsyncRequest::Submit() {
   }
 
   const ServiceKey& service = GetPolarisService(connector_->context_, request_type_);
-  ReturnCode ret_code = connector_->SelectInstance(service, 0, &server_);
-  if (ret_code != kReturnOk) {
-    callback_(ret_code, "select server failed", nullptr);
-    return false;
+  if (service.name_.empty()) {
+    SeedServer& seedServer = connector_->SelectSeed();
+    host_ = seedServer.ip_;
+    port_ = seedServer.port_;
+  } else {
+    ReturnCode ret_code = connector_->SelectInstance(service, 0, &server_);
+    if (ret_code != kReturnOk) {
+      callback_(ret_code, "select server failed", nullptr);
+      return false;
+    }
+    host_ = server_->GetHost();
+    port_ = server_->GetPort();
   }
-
   connector_->async_request_map_[request_id_] = this;  // 记录请求
   // 尝试建立连接
   client_ = new grpc::GrpcClient(reactor_);
-  client_->Connect(server_->GetHost(), server_->GetPort(), GetTimeLeft(),
+  client_->Connect(host_, port_, GetTimeLeft(),
                    std::bind(&AsyncRequest::OnConnect, this, std::placeholders::_1));
   return true;
 }

polaris/plugin/server_connector/grpc_server_connector.h

@@ -175,6 +175,7 @@ class GrpcServerConnector : public ServerConnector, public grpc::StreamCallback<
   virtual ReturnCode SelectInstance(const ServiceKey& service_key, uint32_t timeout, Instance** instance,
                                     bool ignore_half_open = false);
 
+  SeedServer& SelectSeed();
  private:
   friend class AsyncRequest;
   Context* context_;
@@ -235,6 +236,8 @@ class BlockRequest : public grpc::RequestCallback<v1::Response> {
 
  protected:  // protected for test
   Instance* instance_;
+  std::string host_;
+  int port_;
   grpc::GrpcClient* grpc_client_;
 };
 
@@ -299,6 +302,8 @@ class AsyncRequest : public grpc::RequestCallback<v1::Response> {
   uint64_t timeout_;
   PolarisCallback callback_;
   Instance* server_;  // 选择连接的服务器
+  std::string host_;
+  int port_;
   grpc::GrpcClient* client_;
   TimingTaskIter timing_task_;
 };