Merge pull request #21 from lambdaliu/slow_half_open

feat: slow down select half open instance

Author: lambdaliu

polaris/engine/circuit_breaker_executor.cpp

@@ -57,7 +57,7 @@ void CircuitBreakerExecutor::TimingCircuitBreak(CircuitBreakerExecutor* executor
   all_service_contexts.clear();
   // 设置定时任务
   executor->reactor_.AddTimingTask(
-      new TimingFuncTask<CircuitBreakerExecutor>(TimingCircuitBreak, executor, 500));
+      new TimingFuncTask<CircuitBreakerExecutor>(TimingCircuitBreak, executor, 1000));
 }
 
 }  // namespace polaris

polaris/model/model.cpp

@@ -995,6 +995,7 @@ void Service::SetCircuitBreakerData(const CircuitBreakerData& circuit_breaker_da
     impl_->have_half_open_data_ = true;
   } else {
     impl_->have_half_open_data_ = false;
+    impl_->try_half_open_count_ = 20;
   }
 }
 
@@ -1020,6 +1021,19 @@ ReturnCode Service::TryChooseHalfOpenInstance(std::set<Instance*>& instances, In
   if (!impl_->have_half_open_data_ || instances.empty()) {
     return kReturnInstanceNotFound;
   }
+  // 控制释放半开节点的评论，有半开节点以后每20个请求
+  // 且距离上次释放半开节点超过2s就释放1个半开请求
+  if (++impl_->try_half_open_count_ < 20) {
+    return kReturnInstanceNotFound;  // 距离上次释放不足20个正常请求
+  }
+  uint64_t last_half_open_time = impl_->last_half_open_time_.Load();
+  uint64_t current_time        = Time::GetCurrentTimeMs();
+  if (current_time < last_half_open_time + 2000 ||
+      !impl_->last_half_open_time_.Cas(last_half_open_time, current_time)) {
+    return kReturnInstanceNotFound;  // 距离上一次释放半开间隔不足2s
+  }
+  impl_->try_half_open_count_ = 0;
+
   std::set<Instance*>::iterator split_it = instances.begin();
   std::advance(split_it, rand() % instances.size());
   std::set<Instance*>::iterator instance_it;

polaris/model/model_impl.h

@@ -300,6 +300,8 @@ class ServiceImpl {
 
   // 半开优先分配数据
   sync::Mutex half_open_lock_;
+  sync::Atomic<uint64_t> last_half_open_time_;
+  sync::Atomic<int> try_half_open_count_;
   bool have_half_open_data_;
   std::map<std::string, int> half_open_data_;  // 存储半开分配数据
 

polaris/plugin/circuit_breaker/circuit_breaker.h

@@ -35,7 +35,7 @@ static const char kChainEnableKey[]   = "enable";
 static const bool kChainEnableDefault = true;
 
 static const char kChainCheckPeriodKey[]       = "checkPeriod";
-static const uint64_t kChainCheckPeriodDefault = 500;
+static const uint64_t kChainCheckPeriodDefault = 1000;
 
 // 用于传入是否开启探测插件
 static const char kDetectorDisableKey[]   = "detectorDisable";

polaris/plugin/circuit_breaker/error_count.cpp

@@ -140,7 +140,7 @@ ReturnCode ErrorCountCircuitBreaker::TimingCircuitBreak(
           error_count_status.error_count      = 0;
           error_count_status.last_update_time = current_time;
         }
-      } else if (error_count_status.last_update_time + 20 * sleep_window_ <= current_time) {
+      } else if (error_count_status.last_access_time + 100 * sleep_window_ <= current_time) {
         // 兜底：如果访问量一定时间达不到要求，则重新熔断
         if (instances_status->TranslateStatus(it->first, kCircuitBreakerHalfOpen,
                                               kCircuitBreakerOpen)) {

polaris/plugin/circuit_breaker/error_count.h

@@ -25,9 +25,6 @@
 
 namespace polaris {
 
-class Config;
-class Context;
-
 struct ErrorCountStatus {
   CircuitBreakerStatus status;
   int error_count;

polaris/plugin/circuit_breaker/error_rate.cpp

@@ -175,7 +175,7 @@ ReturnCode ErrorRateCircuitBreaker::TimingCircuitBreak(
           error_rate_status.ClearBuckets(metric_num_buckets_);
         }
       } else if (err_req > request_count_after_half_open_ - success_count_after_half_open_ ||
-                 error_rate_status.last_update_time + 20 * sleep_window_ <= current_time) {
+                 error_rate_status.last_access_time + 100 * sleep_window_ <= current_time) {
         // 达到重新熔断条件
         if (instances_status->TranslateStatus(it->first, kCircuitBreakerHalfOpen,
                                               kCircuitBreakerOpen)) {

polaris/plugin/circuit_breaker/error_rate.h

@@ -25,9 +25,6 @@
 
 namespace polaris {
 
-class Config;
-class Context;
-
 struct ErrorRateBucket {
   ErrorRateBucket() : total_count(0), error_count(0), bucket_time(0) {}
   int total_count;

polaris/plugin/health_checker/http_detector.cpp

@@ -36,13 +36,12 @@ ReturnCode HttpHealthChecker::Init(Config* config, Context* /*context*/) {
 
   request_path_ = config->GetStringOrDefault(kHttpRequestPathKey, kHttpRequestPathDefault);
   if (request_path_.empty() || request_path_[0] != '/') {
-    POLARIS_LOG(LOG_ERROR, "outlier detector[%s] config %s invalid", kPluginHttpHealthChecker,
+    POLARIS_LOG(LOG_ERROR, "health checker[%s] config %s invalid", kPluginHttpHealthChecker,
                 kHttpRequestPathKey);
     return kReturnInvalidConfig;
   }
   timeout_ms_ = config->GetMsOrDefault(HealthCheckerConfig::kTimeoutKey,
                                        HealthCheckerConfig::kTimeoutDefault);
-
   return kReturnOk;
 }
 

polaris/plugin/health_checker/tcp_detector.cpp

@@ -36,18 +36,17 @@ ReturnCode TcpHealthChecker::Init(Config* config, Context* /*context*/) {
 
   std::string send_package = config->GetStringOrDefault(kTcpSendPackageKey, kTcpSendPackageDefault);
   if (!send_package.empty() && !Utils::HexStringToBytes(send_package, &send_package_)) {
-    POLARIS_LOG(LOG_ERROR, "outlier detector[%s] config %s hexstring to bytes failed",
+    POLARIS_LOG(LOG_ERROR, "health checker[%s] config %s hexstring to bytes failed",
                 kPluginTcpHealthChecker, kTcpSendPackageKey);
     return kReturnInvalidConfig;
   }
   std::string receive_package =
       config->GetStringOrDefault(kTcpReceivePackageKey, kTcpReceivePackageDefault);
   if (!receive_package.empty() && !Utils::HexStringToBytes(receive_package, &receive_package_)) {
-    POLARIS_LOG(LOG_ERROR, "outlier detector[%s] config %s hexstring to bytes failed",
+    POLARIS_LOG(LOG_ERROR, "health checker[%s] config %s hexstring to bytes failed",
                 kPluginTcpHealthChecker, kTcpReceivePackageKey);
     return kReturnInvalidConfig;
   }
-
   timeout_ms_ = config->GetMsOrDefault(HealthCheckerConfig::kTimeoutKey,
                                        HealthCheckerConfig::kTimeoutDefault);
   return kReturnOk;