优化不健康实例检查逻辑

Author: anjia0532

README.md

@@ -9,7 +9,7 @@ gateway等网关插件的高扩展性
 ### 通过docker运行
 
 ```bash
-docker run anjia0532/discovery-syncer-python:v2.6.9
+docker run anjia0532/discovery-syncer-python:v2.6.12
 ```
 
 特别的，`-c ` 支持配置远端http[s]的地址，比如读取静态资源的，比如读取nacos的

app/handler/discovery.py

@@ -6,8 +6,9 @@
 from fastapi import Response
 from fastapi.params import Path, Body, Query
 
+from core.database import db
 from . import RESP_OK
-from app.model.syncer_model import Registration, RegistrationType, RegistrationStatus
+from app.model.syncer_model import Registration, RegistrationType, RegistrationStatus, DiscoveryInstance
 from app.service.discovery.discovery import Discovery
 from app.service.gateway.gateway import Gateway
 from core.lib.logger import for_handler
@@ -70,6 +71,9 @@ def discovery(discovery_name: Annotated[str, Path(title="discovery_name", descri
                 raise Exception(
                     f"最少存活实例数{alive_num}不满足，总实例数(含之前已下线数量){len(discovery_instances)}，要下线实例数{len(down_hosts)}，剩余在线实例数{len(alive_hosts)}")
         discovery_client.modify_registration(registration, instances=instances)
+        DiscoveryInstance({}).delete_by_instances(
+            [f"{instance.ip}:{instance.port}" for instance in instances if not instance.enabled],
+            db.get_sqla_helper()[1])
     except Exception as e:
         logger.error(f"主动下线上线注册中心的服务失败,discovery_name {discovery_name},registration {registration}",
                      exc_info=e)

app/model/syncer_model.py

@@ -7,6 +7,7 @@
 
 import httpx
 from db_libs.sqla_lib import SqlaReflectHelper
+from funboost import funboost_current_task
 from httpx import TimeoutException
 from pydantic import BaseModel, Field
 from pydantic import model_validator, AliasChoices
@@ -136,8 +137,12 @@ def health_check(self, healthcheck: dict, sqla_helper: SqlaReflectHelper):
         self.set_counts(params, sqla_helper)
         if not success:
             instances = self.get_target_service_all_instance(0, sqla_helper)
+            fct = funboost_current_task()
+            msg_time = None
+            if fct:
+                msg_time = fct.function_result_status.publish_time
             logger.warning(
-                f"健康检查 {self.target_id} {self.service} {schema}{self.instance}{healthcheck.get('uri')} , 实例状态: {json.dumps([d.to_dict_item() for d in instances])}")
+                f"健康检查 {self.target_id} {self.service} {schema}{self.instance}{healthcheck.get('uri')} ,msg_time: {msg_time}, 实例状态: {json.dumps([d.to_dict_item() for d in instances])}")
         if params['status'] != self.status and healthcheck.get("alert", {}).get("url"):
             instances = self.get_target_service_all_instance(0, sqla_helper)
             keyfunc = lambda item: item['status']

app/tasks/common.py

@@ -19,6 +19,8 @@ class FunboostCommonConfig(BoosterParams):
     is_push_to_dlx_queue_when_retry_max_times: bool = True
     # 任务过滤的失效期，为0则永久性过滤任务。例如设置过滤过期时间是1800秒 ， 30分钟前发布过1 + 2 的任务，现在仍然执行，如果是30分钟以内发布过这个任务，则不执行1 + 2
     task_filtering_expire_seconds: int = 0
+    # 消息过期时间，为0永不过期，为10则代表，10秒之前发布的任务如果现在才轮到消费则丢弃任务。
+    msg_expire_senconds: int = 120
     # # 是否对函数入参进行过滤去重.
     do_task_filtering: bool = False
     # 运行时候,是否记录从消息队列获取出来的消息内容

app/tasks/task_syncer.py

@@ -128,6 +128,7 @@ def syncer(target: dict):
         healthcheck = target.get("healthcheck", {})
         if healthcheck:
             try:
+                # syncer 同步任务，只管存或更新
                 DiscoveryInstance({"target_id": target.get('id'), "service": service.name}).save_or_update(
                     discovery_instances, sqla_helper)
                 # 拿到 discovery_instances 和 health_check 里的 unhealthy 比较，将 discovery 的下掉，保留 >= min-hosts
@@ -150,6 +151,8 @@ def syncer(target: dict):
                         # 下线 unhealthy 实例
                         registration = Registration(service_name=service.name, ext_data=target.get("config", {}))
                         discovery_client.modify_registration(registration, unhealthy_instances)
+                    # 删除无效实例
+                    DiscoveryInstance({}).delete_by_instances([d.instance for d in unhealthy], sqla_helper)
             except Exception as e:
                 logger.warning(f"健康检查下线实例失败, {target.get('id', None)} , {service.name}", exc_info=e)