update alb_gateway_unittest

gateway/alb_gateway_unittest.py

@@ -0,0 +1,61 @@
+import logging
+import unittest
+from os import write
+
+from gateway import gateway_config, alb_utils
+from longvideoapi.longvideoapi_config import server_group_id_list
+
+
+class MyTestCase(unittest.TestCase):
+    def test_add_servers_to_server_group(self):
+        alb_client = alb_utils.connect_alb_client(
+            access_key_id=gateway_config.alb_client_params['access_key_id'],
+            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
+            endpoint=gateway_config.alb_client_params['endpoint']
+        )
+        instance_ids = ["i-bp19n839usecekzn2ig9"]
+        alb_utils.add_servers_to_server_group(alb_client, gateway_config.server_group_id_list, instance_ids, weight=100, port=gateway_config.port)
+
+
+    def test_list_server_group_servers(self):
+        alb_client = alb_utils.connect_alb_client(
+            access_key_id=gateway_config.alb_client_params['access_key_id'],
+            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
+            endpoint=gateway_config.alb_client_params['endpoint']
+        )
+        online_instance_ids = alb_utils.list_server_group_servers(alb_client, server_group_id=gateway_config.server_group_id_list[0])
+        logging.info(online_instance_ids)
+
+
+    def test_update_server_group_servers_attribute(self):
+        alb_client = alb_utils.connect_alb_client(
+            access_key_id=gateway_config.alb_client_params['access_key_id'],
+            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
+            endpoint=gateway_config.alb_client_params['endpoint']
+        )
+        instance_id = ["i-bp19n839usecekzn2ig9"]
+        add_weight_list = [(10, 1), (20, 1), (40, 1)]
+        port = gateway_config.port
+        alb_utils.update_server_group_servers_attribute(alb_client=alb_client,
+                                                        server_group_id_list=gateway_config.server_group_id_list,
+                                                        instance_id_list=instance_id,
+                                                        weight_list=add_weight_list,
+                                                        port=port)
+
+    def test_remove_servers_from_server_group(self):
+        alb_client = alb_utils.connect_alb_client(
+            access_key_id=gateway_config.alb_client_params['access_key_id'],
+            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
+            endpoint=gateway_config.alb_client_params['endpoint']
+        )
+        instance_id = ["i-bp19n839usecekzn2ig9"]
+        alb_utils.remove_servers_from_server_group(alb_client,
+                                                   server_group_id_list=gateway_config.server_group_id_list,
+                                                   instance_ids=instance_id,
+                                                   port=gateway_config.port)
+
+
+
+
+if __name__ == '__main__':
+    unittest.main()

gateway/alb_gateway_update.py

@@ -7,7 +7,8 @@ import queue
 import threading
 import time
 import gateway_config
-import utils
+import alb_utils
+from longvideoapi.longvideoapi_config import server_group_id_list
 
 # 从配置文件中获取应用程序名称和容器仓库地址
 apps = gateway_config.apps
@@ -43,7 +44,7 @@ def update(instance_id, port):
     time.sleep(10)  # 等待10秒钟
     global success_count
 
-    ipadd = utils.get_ip_address(ecs_client, instance_id)  # 使用 utils 获取实例的 IP 地址
+    ipadd = alb_utils.get_ip_address(ecs_client, instance_id)  # 使用 utils 获取实例的 IP 地址
     print("服务器信息：" + "%s/%s" % (instance_id, ipadd))
     client = docker.DockerClient(base_url='tcp://%s:2375' % (ipadd), timeout=60)
 
@@ -60,14 +61,22 @@ def update(instance_id, port):
         docker_config = gateway_config.docker_config
         client.login(username=docker_config['username'], password=docker_config['password'],
                      registry=docker_config['registry'])
-        client.containers.run(registry.format(apps, version), detach=True, cap_add='SYS_PTRACE', network_mode='host', name=apps,
+        client.containers.run(registry.format(apps, version),
+                              detach=True,
+                              cap_add='SYS_PTRACE',
+                              network_mode='host',
+                              name=apps,
                               volumes={'/datalog/': {'bind': '/datalog/', 'mode': 'rw'}})
         print("开始健康检查")
         checkHealth(ipadd)
         print("%s ：权重修改中......" % (ipadd))
         weight_list = [(10, 5), (20, 5), (40, 5), (60, 5), (80, 5), (100, 5)]
         # weight_list = [(10, 10), (20, 10), (40, 10), (60, 10), (80, 10), (100, 10)]
-        utils.update_server_group_servers_attribute(alb_client, gateway_config.server_group_id_list, instance_id_list=[instance_id], weight_list=weight_list, port=port)
+        alb_utils.update_server_group_servers_attribute(alb_client,
+                                                        server_group_id_list=gateway_config.server_group_id_list,
+                                                        instance_id_list=[instance_id],
+                                                        weight_list=weight_list,
+                                                        port=port)
         success_count += 1
         print("更新进度" + "%s/%s" % (success_count, total))
     except Exception as e:
@@ -78,7 +87,7 @@ def update(instance_id, port):
 def pull_image():
     """从镜像仓库中拉取指定版本的镜像"""
     instanceId = q1.get()
-    ipaddr = utils.get_ip_address(ecs_client, instanceId)
+    ipaddr = alb_utils.get_ip_address(ecs_client, instanceId)
     cd_url = "tcp://{}:2375".format(ipaddr)
     client = docker.DockerClient(base_url=cd_url, timeout=30)
 
@@ -93,11 +102,11 @@ def pull_image():
 
 if __name__ == '__main__':
     # 初始化 ECS 客户端
-    ecs_client = utils.connect_client(access_key_id=gateway_config.ecs_client_params['access_key_id'],
+    ecs_client = alb_utils.connect_client(access_key_id=gateway_config.ecs_client_params['access_key_id'],
                                       access_key_secret=gateway_config.ecs_client_params['access_key_secret'],
                                       region_id=gateway_config.ecs_client_params['region_id'])
     # 初始化 ALB 客户端
-    alb_client = utils.connect_alb_client(gateway_config.alb_client_params['access_key_id'],
+    alb_client = alb_utils.connect_alb_client(gateway_config.alb_client_params['access_key_id'],
                                           gateway_config.alb_client_params['access_key_secret'],
                                           endpoint=gateway_config.alb_client_params['endpoint']
                                           )
@@ -109,7 +118,7 @@ if __name__ == '__main__':
     q1 = queue.Queue()
 
     # 获取 ALB 下服务器组的实例 ID
-    res = utils.list_server_group_servers(alb_client=alb_client, server_group_id=gateway_config.server_group_id_list[0])
+    res = alb_utils.list_server_group_servers(alb_client=alb_client, server_group_id=gateway_config.server_group_id_list[0])
     total += len(res)
     print(f"获取 ALB 下服务器组的实例 ID = {res} total = {total}")
     InstanceIDs.extend(res)
@@ -130,7 +139,9 @@ if __name__ == '__main__':
 
     # 更新每个实例
     for instanceID in InstanceIDs:
-        for server_group_id in gateway_config.server_group_id_list:
-            utils.update_server_group_server_weight(alb_client, server_group_id, instanceID,
-                                                    weight=0, port=gateway_config.port)  # 设置初始权重为0
+        alb_utils.update_server_group_server_weight(alb_client=alb_client,
+                                                    server_group_id_list=server_group_id_list,
+                                                    instance_id_list=instanceID,
+                                                    weight=0,
+                                                    port=gateway_config.port)
         update(instanceID, port=gateway_config.port)

gateway/gateway_unittest.py

@@ -1,8 +0,0 @@
-import unittest
-
-class MyTestCase(unittest.TestCase):
-    def test_something(self):
-        self.assertEqual(True, False)  # add assertion here
-
-if __name__ == '__main__':
-    unittest.main()