Method check_healthy

LoadBalancing.py:29–56

从本地健康检查服务获取API健康状态数据 Returns: dict: 包含时间戳、检查间隔和健康数据的字典

()

class Harness_localAPI:
    @staticmethod
    def check_healthy():
        """从本地健康检查服务获取API健康状态数据
        
        Returns:
            dict: 包含时间戳、检查间隔和健康数据的字典
        """
        logger = logging.getLogger(__name__)
        # 防止日志向上传播，避免重复打印
        logger.propagate = False
        try:
            # 调用本地健康检查服务API
            logger.info("正在调用健康检查服务API获取健康状态数据")
            response = requests.get(API_HEALTH_CHECK_URL, timeout=INNER_TIMEOUT)
            if response.status_code == 200:
                result = response.json()
                # 将字符串键转换为元组键
                if "data" in result:
                    result["data"] = {tuple(key.split("|")): [np.nan if v is None else v for v in value] for key, value in result["data"].items()}
                logger.info("成功获取健康状态数据")
                return result
            else:
                # 如果API调用失败，返回空数据结构和当前时间
                logger.error(f"健康检查API返回错误状态码: {response.status_code}")
                return {"timestamp": datetime.now().isoformat(), "check_timer_span": 15, "data": {}}
        except Exception as e:
            # 出现异常时记录错误并返回空数据结构
            logger.error(f"获取健康状态数据时出错: {str(e)}")
            return {"timestamp": datetime.now().isoformat(), "check_timer_span": 15, "data": {}}

    @staticmethod
    def get_api_key(source_name):