distribution_update_b.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351
  1. import asyncio
  2. import sys
  3. import time
  4. import requests
  5. import alb_utils
  6. import logging
  7. import os
  8. import docker
  9. import distribution_config
  10. from concurrent.futures import ThreadPoolExecutor
  11. health_instances = []
  12. ess_instances = []
  13. remove_container_instances = []
  14. def server_health_check(ecs_client, instance_id):
  15. """
  16. 服务健康检查
  17. :param ecs_client: 客户端连接
  18. :param instance_id: instanceId
  19. :return:
  20. """
  21. global health_instances
  22. ip_address = alb_utils.get_ip_address(ecs_client=ecs_client, instance_id=instance_id)
  23. while True:
  24. health_check_url = f"http://{ip_address}:8080/healthcheck"
  25. try:
  26. http_code = requests.get(health_check_url).status_code
  27. except:
  28. logging.info(f"images is downloading ip:{ip_address}")
  29. http_code = 0
  30. if http_code == 200:
  31. health_instances.append((instance_id, ip_address))
  32. logging.info(f"health check success, instance: {instance_id}/{ip_address}")
  33. break
  34. else:
  35. time.sleep(10)
  36. async def ess_instance(ecs_client, alb_client, ess_count, max_workers, version, port):
  37. """
  38. 扩容机器并运行新服务
  39. :param ecs_client: 购买机器客户端连接
  40. :param alb_client: 修改负载均衡权限
  41. :param ess_count: 扩容数量
  42. :param max_workers: 线程数
  43. :param version: 版本标记
  44. :param port: 后端服务器使用的端口
  45. :return:
  46. """
  47. # 1. 购买机器并启动
  48. ess_instance_ids = alb_utils.create_multiple_instances(
  49. amount=ess_count,
  50. ecs_client=ecs_client,
  51. **distribution_config.instance_config_b,
  52. )
  53. time.sleep(60)
  54. # 2. 发送启动脚本到机器上
  55. alb_utils.send_file_to_ecs(ecs_client=ecs_client, instance_id_list=ess_instance_ids, **distribution_config.start_sh)
  56. logging.info(f"send start shell file finished, count: {len(ess_instance_ids)} instances: {ess_instance_ids}")
  57. # 3. 启动服务
  58. server_start_sh = os.path.join(distribution_config.start_sh['target_dir'], distribution_config.start_sh['name'])
  59. server_start_commend = f"sh {server_start_sh} {version}"
  60. alb_utils.run_command(ecs_client=ecs_client, instance_ids=ess_instance_ids, command=server_start_commend)
  61. logging.info(f"start server finish, count: {len(ess_instance_ids)} instances: {ess_instance_ids}")
  62. # 4. 异步探活
  63. global health_instances
  64. health_instances = []
  65. loop = asyncio.get_running_loop()
  66. executor = ThreadPoolExecutor(max_workers=max_workers)
  67. tasks = [
  68. loop.run_in_executor(executor, server_health_check, *args) for args in
  69. [(ecs_client, instance_id) for instance_id in ess_instance_ids]
  70. ]
  71. await asyncio.wait(tasks)
  72. logging.info(f"health instances count: {len(health_instances)}, instances: {health_instances}")
  73. # 5. 挂载流量
  74. if len(health_instances) == len(ess_instance_ids):
  75. # 所有机器探活成功
  76. time.sleep(20)
  77. health_instance_ids = [instance_id for instance_id, _ in health_instances]
  78. alb_utils.add_servers_to_server_group(alb_client, distribution_config.server_group_id_list, health_instance_ids, weight=0, port=port)
  79. logging.info(f"Successfully added count: {len(health_instance_ids)} health_instance_ids {health_instance_ids} to server groups {distribution_config.server_group_id_list}.")
  80. time.sleep(20)
  81. logging.info(f"start update weight count: {len(health_instance_ids)} instances: {health_instance_ids} server groups: {distribution_config.server_group_id_list}.")
  82. add_weight_list = [(10, 5), (20, 5), (40, 5), (60, 5), (80, 5), (100, 5)]
  83. alb_utils.update_server_group_servers_attribute(alb_client,
  84. server_group_id_list=distribution_config.server_group_id_list,
  85. instance_id_list=health_instance_ids,
  86. weight_list=add_weight_list,
  87. port=port)
  88. global ess_instances
  89. ess_instances.extend(health_instance_ids)
  90. logging.info(f"ess count: {ess_count}, "
  91. f"create count: {len(ess_instance_ids)}, "
  92. f"finished count: {len(health_instance_ids)}")
  93. else:
  94. logging.info(f"ess count: {ess_count}, "
  95. f"create count: {len(ess_instance_ids)}, "
  96. f"health count: {len(health_instances)}")
  97. sys.exit()
  98. def remove_container_image(ecs_client, instance_id, container_name_list):
  99. """
  100. 移除旧容器并删除旧镜像
  101. :param ecs_client: 客户端连接
  102. :param instance_id: instanceId type-string
  103. :param container_name: 容器名称 type-string
  104. :return:
  105. """
  106. ip_address = alb_utils.get_ip_address(ecs_client=ecs_client, instance_id=instance_id)
  107. logging.info(f"服务器信息:{instance_id}/{ip_address}")
  108. client = docker.DockerClient(base_url=f'tcp://{ip_address}:2375', timeout=60)
  109. # 移除旧的容器
  110. container_remove_retry = 3
  111. i = 0
  112. while True:
  113. if i >= container_remove_retry:
  114. logging.error(f"容器不存在或者无法删除当前容器, instance = {instance_id}/{ip_address}")
  115. sys.exit()
  116. try:
  117. flag = False
  118. for container_name in container_name_list:
  119. try:
  120. container_id = client.containers.get(container_name)
  121. container_id.remove(force=True)
  122. flag = True
  123. break
  124. except:
  125. continue
  126. if flag:
  127. break
  128. except Exception as e:
  129. i += 1
  130. # 删除旧镜像
  131. images_remove_retry = 3
  132. j = 0
  133. while True:
  134. if j >= images_remove_retry:
  135. logging.error(f"镜像不存在,无法获取到镜像ID, instance = {instance_id}/{ip_address}")
  136. sys.exit()
  137. try:
  138. images = client.images.list()
  139. for image in images:
  140. client.images.remove(force=True, image=image.tags[0])
  141. time.sleep(2)
  142. global remove_container_instances
  143. remove_container_instances.append(instance_id)
  144. break
  145. except Exception as e:
  146. i += 1
  147. async def update_instance(ecs_client, alb_client, instance_ids, max_workers, version, port):
  148. """
  149. 线上机器更新
  150. :param ecs_client:
  151. :param alb_client: alb客户端连接
  152. :param instance_ids: instanceId type-list
  153. :param max_workers:
  154. :param version: 版本标记
  155. :param port: 后端服务器使用的端口
  156. :return:
  157. """
  158. sub_index = len(instance_ids)//2
  159. instance_ids_group = [instance_ids[:sub_index], instance_ids[sub_index:]]
  160. update_finished_count = 0
  161. for instance_id_list in instance_ids_group:
  162. logging.info(f"update instances: {instance_id_list}")
  163. # 1. 摘流量
  164. alb_utils.update_server_group_servers_attribute(alb_client=alb_client,
  165. server_group_id_list=distribution_config.server_group_id_list,
  166. instance_id_list=instance_id_list,
  167. weight_list=[(0, 20)],
  168. port=port)
  169. logging.info(f"set weight with 0 finished, count: {len(instance_id_list)} instances: {instance_id_list}")
  170. # 2. 异步移除旧容器并删除旧镜像
  171. global remove_container_instances
  172. remove_container_instances = []
  173. container_name_list = ['distribution']
  174. loop = asyncio.get_running_loop()
  175. executor = ThreadPoolExecutor(max_workers=max_workers)
  176. tasks = [
  177. loop.run_in_executor(executor, remove_container_image, *args) for args in
  178. [(ecs_client, instance_id, container_name_list) for instance_id in instance_id_list]
  179. ]
  180. await asyncio.wait(tasks)
  181. logging.info(f"remove container & images finished, instances: {remove_container_instances},"
  182. f" count: {len(remove_container_instances)}")
  183. if len(remove_container_instances) < len(instance_id_list):
  184. logging.error(f"remove container image failed| "
  185. f"request count: {len(instance_id_list)}, removed count: {len(remove_container_instances)}")
  186. sys.exit()
  187. # 3. 发送启动脚本到机器上
  188. alb_utils.send_file_to_ecs(ecs_client=ecs_client, instance_id_list=instance_id_list, **distribution_config.start_sh)
  189. logging.info(f"send start shell file finished, instances: {instance_id_list}, count: {len(instance_id_list)}")
  190. # 4. 启动服务
  191. server_start_sh = os.path.join(distribution_config.start_sh['target_dir'], distribution_config.start_sh['name'])
  192. server_start_commend = f"sh {server_start_sh} {version}"
  193. alb_utils.run_command(ecs_client=ecs_client, instance_ids=instance_id_list, command=server_start_commend)
  194. # 5. 异步探活
  195. global health_instances
  196. health_instances = []
  197. loop = asyncio.get_running_loop()
  198. executor = ThreadPoolExecutor(max_workers=max_workers)
  199. tasks = [
  200. loop.run_in_executor(executor, server_health_check, *args) for args in
  201. [(ecs_client, instance_id) for instance_id in instance_id_list]
  202. ]
  203. await asyncio.wait(tasks)
  204. logging.info(f"health instances: {health_instances}, count: {len(health_instances)}")
  205. # 6. 挂载流量
  206. if len(health_instances) == len(instance_id_list):
  207. # 所有机器探活成功
  208. time.sleep(10)
  209. health_instance_ids = [instance_id for instance_id, _ in health_instances]
  210. add_weight_list = [(10, 5), (20, 5), (40, 5), (60, 5), (80, 5), (100, 5)]
  211. # add_weight_list = [(10, 10), (20, 10), (40, 10), (60, 10), (80, 10), (100, 10)]
  212. alb_utils.update_server_group_servers_attribute(alb_client,
  213. server_group_id_list=distribution_config.server_group_id_list,
  214. instance_id_list=health_instance_ids,
  215. weight_list=add_weight_list,
  216. port=port)
  217. logging.info(f"finished instances: {health_instances}, count: {len(health_instances)}")
  218. update_finished_count += len(health_instances)
  219. logging.info(f"update finished: {update_finished_count}/{len(instance_ids)}")
  220. else:
  221. logging.info(f"health instances: {health_instances}, count: {len(health_instances)}")
  222. sys.exit()
  223. def remove_instances(ecs_client, alb_client, instance_ids, port):
  224. """
  225. 停止并释放机器
  226. :param ecs_client:
  227. :param alb_client:
  228. :param instance_ids: instanceId type-list
  229. :param port: 后端服务器使用的端口
  230. :return: None
  231. """
  232. # 1. 摘流量
  233. weight_list = [(0, 20)] # 设置权重为0,等待20秒
  234. try:
  235. alb_utils.update_server_group_servers_attribute(alb_client=alb_client,
  236. server_group_id_list=distribution_config.server_group_id_list,
  237. instance_id_list=instance_ids,
  238. weight_list=weight_list,
  239. port=port)
  240. except Exception as e:
  241. logging.error(f"Failed to set instance weight: {e}")
  242. sys.exit()
  243. time.sleep(10)
  244. # 2.移除alb
  245. # for server_group_id in distribution_config.server_group_id_list:
  246. try:
  247. alb_utils.remove_servers_from_server_group(alb_client=alb_client, server_group_id_list=distribution_config.server_group_id_list,
  248. instance_ids=instance_ids, port=port)
  249. logging.info(f"Successfully removed instances: {instance_ids} from server group {distribution_config.server_group_id_list}.")
  250. except Exception as e:
  251. logging.error(f"Failed to remove instances: {instance_ids} from server group {distribution_config.server_group_id_list}: {e}")
  252. logging.info(f"Remove from ALB finished, instances: {instance_ids}")
  253. # 3. 停止机器
  254. alb_utils.stop_instances(ecs_client=ecs_client, instance_ids=instance_ids)
  255. # stop_response = alb_utils.stop_instances(ecs_client=ecs_client, instance_ids=instance_ids)
  256. # if stop_response.get('Code') is None:
  257. # logging.info(f"Instances stop finished, instances: {instance_ids}")
  258. # else:
  259. # logging.error(f"Failed to stop instances: {stop_response}")
  260. # sys.exit()
  261. # 4. 判断机器运行状态是否为Stopped
  262. # while True:
  263. stopped_instances = alb_utils.get_instances_status(ecs_client=ecs_client, instance_ids=instance_ids)
  264. # response = alb_utils.get_instances_status(ecs_client=ecs_client, instance_ids=instance_ids)
  265. # if response.get('Code') is None:
  266. # instances_list = response.get('InstanceStatuses').get('InstanceStatus')
  267. # stopped_instances = [instance.get('InstanceId') for instance in instances_list if
  268. # instance.get('Status') == 'Stopped']
  269. # if len(stopped_instances) == len(instance_ids):
  270. # logging.info(f"Instances stopped status set success, instances: {stopped_instances}")
  271. # break
  272. # else:
  273. # logging.info(f"Stopped instances count = {len(stopped_instances)}, instances: {stopped_instances}")
  274. # time.sleep(5)
  275. # else:
  276. # logging.error(response)
  277. # sys.exit()
  278. # 5. 释放机器
  279. alb_utils.release_instances(ecs_client=ecs_client, instance_ids=stopped_instances)
  280. # release_response = alb_utils.release_instances(ecs_client=ecs_client, instance_ids=stopped_instances)
  281. # if release_response.get('Code') is None:
  282. # logging.info(f"Release instances finished, instances: {stopped_instances}")
  283. # else:
  284. # logging.error(f"Release instances fail!!!")
  285. # sys.exit()
  286. def main():
  287. try:
  288. version = sys.argv[1]
  289. alb_client = alb_utils.connect_alb_client(
  290. access_key_id=distribution_config.alb_client_params['access_key_id'],
  291. access_key_secret=distribution_config.alb_client_params['access_key_secret'],
  292. endpoint=distribution_config.alb_client_params['endpoint']
  293. )
  294. ecs_client = alb_utils.connect_client(access_key_id=distribution_config.ecs_client_params['access_key_id'],
  295. access_key_secret=distribution_config.ecs_client_params['access_key_secret'],
  296. region_id=distribution_config.ecs_client_params['region_id'])
  297. # 1. 获取alb下所有机器
  298. online_instance_ids = alb_utils.list_server_group_servers(alb_client=alb_client, server_group_id=distribution_config.server_group_id_list[0])
  299. online_instance_count = len(online_instance_ids)
  300. logging.info(f"online instance count: {online_instance_count} instance_ids: {online_instance_ids}")
  301. # 2. 扩容机器并启动新服务 扩容数量:线上机器数量//2
  302. logging.info(f"ess instances start ...")
  303. ess_instance_count = online_instance_count // 2
  304. logging.info(f"ess instance count: {ess_instance_count}")
  305. asyncio.run(ess_instance(ecs_client=ecs_client, alb_client=alb_client,
  306. ess_count=ess_instance_count, max_workers=2, version=version, port=distribution_config.port))
  307. logging.info(f"ess instances end!")
  308. # 3. 原有机器进行更新
  309. logging.info(f"update online instances start ...")
  310. asyncio.run(update_instance(ecs_client=ecs_client, alb_client=alb_client,
  311. instance_ids=online_instance_ids, max_workers=8, version=version, port=distribution_config.port))
  312. logging.info(f"update online instances end!")
  313. # 4. 停止并释放扩容机器
  314. logging.info(f"stop & release instances start ...")
  315. remove_instances(ecs_client=ecs_client, alb_client=alb_client, instance_ids=ess_instances, port=distribution_config.port)
  316. logging.info(f"stop & release instances end!")
  317. except Exception as e:
  318. logging.error(e)
  319. sys.exit()
  320. if __name__ == '__main__':
  321. main()