fix service instances expired failed

lambdaliu · lambdaliu · commit a5f096bfb213 · 2021-10-05T22:06:02.000+08:00
diff --git a/include/polaris/plugin.h b/include/polaris/plugin.h
@@ -285,6 +285,10 @@ class LocalRegistry : public Plugin {
   virtual ReturnCode UpdateSetCircuitBreakerData(
       const ServiceKey& service_key, const CircuitBreakUnhealthySetsData& unhealthy_sets) = 0;
 
+  virtual ReturnCode GetCircuitBreakerInstances(const ServiceKey& service_key,
+                                                ServiceData*& service_data,
+                                                std::vector<Instance*>& open_instances) = 0;
+
   /// @brief 更新服务实例状态，properties存放的是状态值，当前支持2个key
   ///
   /// 1. ReadyToServe: 故障熔断标识，true or false
diff --git a/polaris/cache/rcu_map.h b/polaris/cache/rcu_map.h
@@ -78,7 +78,7 @@ class RcuMap {
   ~RcuMap();
 
   /// @brief 根据Key获取指向Value的指针，key不存在返回NULL
-  Value* Get(const Key& key);
+  Value* Get(const Key& key, bool update_access_time = true);
 
   /// @brief 更新Key对应的Value
   /// 如果key对应的value已存在，则将旧的value加入待释放列表，内部线程会延迟一定时间释放
@@ -166,20 +166,24 @@ RcuMap<Key, Value>::~RcuMap() {
 }
 
 template <typename Key, typename Value>
-Value* RcuMap<Key, Value>::Get(const Key& key) {
+Value* RcuMap<Key, Value>::Get(const Key& key, bool update_access_time) {
   // 查询read map，获取结果
   Value* read_result             = NULL;
   InnerMap* current_read         = read_map_;
   typename InnerMap::iterator it = current_read->find(key);
   if (it != current_read->end()) {  // MapValue包含的value指针在整个过程中是可能改变的
-    it->second->used_time_ = Time::GetCurrentTimeMs();
-    read_result            = it->second->value_;
+    if (update_access_time) {
+      it->second->used_time_ = Time::GetCurrentTimeMs();
+    }
+    read_result = it->second->value_;
   } else {
     // 从read map未读到数据，则加锁进行后续操作
     sync::MutexGuard mutex_guard(dirty_lock_);
     if ((it = dirty_map_->find(key)) != dirty_map_->end()) {
-      it->second->used_time_ = Time::GetCurrentTimeMs();
-      read_result            = it->second->value_;
+      if (update_access_time) {
+        it->second->used_time_ = Time::GetCurrentTimeMs();
+      }
+      read_result = it->second->value_;
       if (read_map_ == current_read) {
         miss_time_++;  // 记录read map读失败，dirty map读成功次数
       }
diff --git a/polaris/plugin/health_checker/health_checker.cpp b/polaris/plugin/health_checker/health_checker.cpp
@@ -138,38 +138,32 @@ ReturnCode HealthCheckerChainImpl::DetectInstance(CircuitBreakerChain& circuit_b
   }
 
   ServiceData* service_data = NULL;
-  local_registry_->GetServiceDataWithRef(service_key_, kServiceDataInstances, service_data);
-  if (service_data == NULL) {
+  std::vector<Instance*> health_check_instances;
+  if (local_registry_->GetCircuitBreakerInstances(service_key_, service_data,
+                                                  health_check_instances) != kReturnOk) {
     return kReturnOk;
   }
-  Service* service = service_data->GetService();
+
   ServiceInstances service_instances(service_data);
   std::map<std::string, Instance*>& instance_map = service_instances.GetInstances();
-  std::set<std::string> target_health_check_instances;
 
   if (when_ == HealthCheckerConfig::kChainWhenAlways) {
+    health_check_instances.clear();
     // 健康检查设置为always, 则探测所有非隔离实例
     for (std::map<std::string, Instance*>::iterator instance_iter = instance_map.begin();
          instance_iter != instance_map.end(); ++instance_iter) {
       if (!instance_iter->second->isIsolate()) {
-        target_health_check_instances.insert(instance_iter->first);
+        health_check_instances.push_back(instance_iter->second);
       }
     }
-  } else if (when_ == HealthCheckerConfig::kChainWhenOnRecover) {
-    // 健康检查设置为on_recover, 则探测半开实例
-    target_health_check_instances = service->GetCircuitBreakerOpenInstances();
+  } else if (when_ != HealthCheckerConfig::kChainWhenOnRecover) {
+    // 健康检查设置不为on_recover, 则探测半开实例
+    health_check_instances.clear();
   }
-  for (std::set<std::string>::iterator it = target_health_check_instances.begin();
-       it != target_health_check_instances.end(); ++it) {
-    const std::string& instance_id                  = *it;
-    std::map<std::string, Instance*>::iterator iter = instance_map.find(instance_id);
-    if (iter == instance_map.end()) {
-      POLARIS_LOG(LOG_INFO, "The health checker of service[%s/%s] getting instance[%s] failed",
-                  service_key_.namespace_.c_str(), service_key_.name_.c_str(), instance_id.c_str());
-      continue;
-    }
+
+  for (std::size_t i = 0; i < health_check_instances.size(); ++i) {
     bool is_detect_success = false;
-    Instance* instance     = iter->second;
+    Instance* instance     = health_check_instances[i];
     for (std::size_t i = 0; i < health_checker_list_.size(); ++i) {
       HealthChecker*& detector = health_checker_list_[i];
       DetectResult detector_result;
@@ -195,15 +189,16 @@ ReturnCode HealthCheckerChainImpl::DetectInstance(CircuitBreakerChain& circuit_b
     // 探活插件成功，则将熔断实例置为半开状态，其他实例状态不变
     // 探活插件失败，则将健康实例置为熔断状态，其他实例状态不变
     if (is_detect_success) {
-      circuit_breaker_chain.TranslateStatus(instance_id, kCircuitBreakerOpen,
+      circuit_breaker_chain.TranslateStatus(instance->GetId(), kCircuitBreakerOpen,
                                             kCircuitBreakerHalfOpen);
       POLARIS_LOG(LOG_INFO,
                   "service[%s/%s] getting instance[%s-%s:%d] detectoring success, change to "
                   "half-open status",
                   service_key_.namespace_.c_str(), service_key_.name_.c_str(),
                   instance->GetId().c_str(), instance->GetHost().c_str(), instance->GetPort());
     } else {
-      circuit_breaker_chain.TranslateStatus(instance_id, kCircuitBreakerClose, kCircuitBreakerOpen);
+      circuit_breaker_chain.TranslateStatus(instance->GetId(), kCircuitBreakerClose,
+                                            kCircuitBreakerOpen);
       POLARIS_LOG(LOG_INFO,
                   "service[%s/%s] getting instance[%s-%s:%d] detectoring failed, change to "
                   "open status",
diff --git a/polaris/plugin/local_registry/local_registry.cpp b/polaris/plugin/local_registry/local_registry.cpp
@@ -175,6 +175,7 @@ void InMemoryRegistry::CheckExpireServiceData(uint64_t min_access_time,
     pthread_rwlock_unlock(&notify_rwlock_);
     if (service_data_type == kServiceDataInstances) {  // 清除实例数据时对应的服务级别插件也删除
       context_impl->DeleteServiceContext(expired_services[i]);
+      DeleteServiceInLock(expired_services[i]);
     }
   }
 }
@@ -308,6 +309,7 @@ ReturnCode InMemoryRegistry::UpdateServiceData(const ServiceKey& service_key,
   }
   if (service_data == NULL) {  // Server Connector反注册Handler触发更新为NULL
     if (data_type == kServiceDataInstances) {  // 删除服务实例数据时，同时删除服务
+      context_impl->DeleteServiceContext(service_key);
       DeleteServiceInLock(service_key);
     }
     context_impl->GetServiceRecord()->ServiceDataDelete(service_key,
@@ -389,4 +391,37 @@ ReturnCode InMemoryRegistry::UpdateSetCircuitBreakerData(
   return service->WriteCircuitBreakerUnhealthySets(unhealthy_sets);
 }
 
+ReturnCode InMemoryRegistry::GetCircuitBreakerInstances(const ServiceKey& service_key,
+                                                        ServiceData*& service_data,
+                                                        std::vector<Instance*>& open_instances) {
+  service_data = service_instances_data_.Get(service_key, false);
+  if (service_data == NULL) {
+    return kReturnServiceNotFound;
+  }
+  if (service_data->GetDataStatus() < kDataIsSyncing) {
+    service_data->DecrementRef();
+    return kReturnServiceNotFound;
+  }
+  Service* service = service_data->GetService();
+  ServiceInstances service_instances(service_data);
+  std::map<std::string, Instance*>& instance_map      = service_instances.GetInstances();
+  std::set<std::string> circuit_breaker_open_instance = service->GetCircuitBreakerOpenInstances();
+  for (std::set<std::string>::iterator it = circuit_breaker_open_instance.begin();
+       it != circuit_breaker_open_instance.end(); ++it) {
+    const std::string& instance_id                  = *it;
+    std::map<std::string, Instance*>::iterator iter = instance_map.find(instance_id);
+    if (iter == instance_map.end()) {
+      POLARIS_LOG(LOG_INFO, "The outlier detector of service[%s/%s] getting instance[%s] failed",
+                  service_key.namespace_.c_str(), service_key.name_.c_str(), instance_id.c_str());
+      continue;
+    }
+    open_instances.push_back(iter->second);
+  }
+  if (open_instances.empty()) {
+    return kReturnInstanceNotFound;
+  }
+  service_data->IncrementRef();
+  return kReturnOk;
+}
+
 }  // namespace polaris
diff --git a/polaris/plugin/local_registry/local_registry.h b/polaris/plugin/local_registry/local_registry.h
@@ -95,6 +95,10 @@ class InMemoryRegistry : public LocalRegistry {
   virtual ReturnCode UpdateSetCircuitBreakerData(
       const ServiceKey& service_key, const CircuitBreakUnhealthySetsData& unhealthy_sets);
 
+  virtual ReturnCode GetCircuitBreakerInstances(const ServiceKey& service_key,
+                                                ServiceData*& service_data,
+                                                std::vector<Instance*>& open_instances);
+
   virtual ReturnCode UpdateDynamicWeight(const ServiceKey& service_key,
                                          const DynamicWeightData& dynamic_weight_data);
 
diff --git a/test/mock/mock_local_registry.h b/test/mock/mock_local_registry.h
@@ -59,6 +59,10 @@ class MockLocalRegistry : public LocalRegistry {
                ReturnCode(const ServiceKey &service_key,
                           const CircuitBreakUnhealthySetsData &cb_unhealthy_set_data));
 
+  MOCK_METHOD3(GetCircuitBreakerInstances,
+               ReturnCode(const ServiceKey &service_key, ServiceData *&service_data,
+                          std::vector<Instance *> &open_instances));
+
   MOCK_METHOD2(UpdateDynamicWeight, ReturnCode(const ServiceKey &service_key,
                                                const DynamicWeightData &dynamic_weight_data));
 
