import sys import os import asyncio import logging import time import requests import utils import longvideo_config from concurrent.futures import ThreadPoolExecutor health_instances = [] def longvideo_health_check(client, instance_id, max_wait_time=None): """ 服务健康检查 :param client: 客户端连接 :param instance_id: instanceId :param max_wait_time: 最长等待时间,单位:s :return: """ global health_instances start_time = time.time() ip_address = utils.get_ip_address(client=client, instance_id=instance_id) while True: health_check_url = f"http://{ip_address}:8080/longvideoapi/test" try: http_code = requests.get(health_check_url).status_code except: logging.info(f"images is downloading ip: {ip_address}") http_code = 0 if http_code == 200: health_instances.append((instance_id, ip_address)) logging.info(f"health check success, instance: {instance_id}/{ip_address}") break elif max_wait_time is not None: now = time.time() if (now - start_time) >= max_wait_time: logging.info(f"health check error, instance: {instance_id}/{ip_address}") break else: time.sleep(10) else: time.sleep(10) async def ess_instance(create_client, slb_client, ess_count, max_workers): """ 扩容机器并运行新服务 :param create_client: 购买机器客户端连接 :param slb_client: 修改负载均衡权限 :param ess_count: 扩容数量 :param max_workers: 线程数 :return: """ # 1. 购买机器并启动 ess_instance_ids = utils.create_multiple_instances( amount=ess_count, client=create_client, **longvideo_config.instance_config_i, ) time.sleep(60) # 2. 发送启动脚本到机器上 utils.send_file_to_ecs(client=create_client, instance_id_list=ess_instance_ids, **longvideo_config.start_sh) logging.info(f"send start shell file finished, instances: {ess_instance_ids}") # 3. 启动服务 start_sh_param = "latest" server_start_sh = os.path.join(longvideo_config.start_sh['target_dir'], longvideo_config.start_sh['name']) server_start_commend = f"sh {server_start_sh} {start_sh_param}" utils.run_command(client=create_client, instance_ids=ess_instance_ids, command=server_start_commend) # 4. 异步探活 global health_instances health_instances = [] max_wait_time = 180 loop = asyncio.get_running_loop() executor = ThreadPoolExecutor(max_workers=max_workers) tasks = [ loop.run_in_executor(executor, longvideo_health_check, *args) for args in [(slb_client, instance_id, max_wait_time) for instance_id in ess_instance_ids] ] await asyncio.wait(tasks) logging.info(f"health instances count: {len(health_instances)}, {health_instances}") # 5. 挂载流量 if len(health_instances) > 0: # 所有机器探活成功 time.sleep(20) utils.add_backend_servers_with_slbs(client=slb_client, slb_id_list=longvideo_config.slb_id_list, instances=health_instances) add_weight_list = [(10, 5), (20, 5), (40, 5), (60, 5), (80, 5), (100, 5)] health_instance_ids = [instance_id for instance_id, _ in health_instances] utils.set_instance_weight_process_with_slbs(client=slb_client, slb_id_list=longvideo_config.slb_id_list, instance_id_list=health_instance_ids, weight_list=add_weight_list) logging.info(f"ess count: {ess_count}, " f"create count: {len(ess_instance_ids)}, " f"finished count: {len(health_instance_ids)}") else: logging.info(f"ess count: {ess_count}, " f"create count: {len(ess_instance_ids)}, " f"health count: {len(health_instances)}") sys.exit() def main(): try: slb_client = utils.connect_client(access_key_id=longvideo_config.slb_client_params['access_key_id'], access_key_secret=longvideo_config.slb_client_params['access_key_secret'], region_id=longvideo_config.slb_client_params['region_id']) create_client = utils.connect_client(access_key_id=longvideo_config.create_client_params['access_key_id'], access_key_secret=longvideo_config.create_client_params['access_key_secret'], region_id=longvideo_config.create_client_params['region_id']) # 获取批量创建ECS实例的数量 ess_instance_count = int(sys.argv[1]) # 扩容机器并启动服务 logging.info(f"ess instances start ...") logging.info(f"ess instance count: {ess_instance_count}") asyncio.run(ess_instance(create_client=create_client, slb_client=slb_client, ess_count=ess_instance_count, max_workers=2)) logging.info(f"ess instances end!") except Exception as e: logging.error(e) if __name__ == '__main__': main()