a2s_monitor/utils/ecs.py

# coding:utf-8
import json
import logging
import time
from aliyunsdkcore import client
from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest
from aliyunsdkecs.request.v20140526.RunInstancesRequest import RunInstancesRequest
from aliyunsdkecs.request.v20140526.DeleteInstanceRequest import DeleteInstanceRequest


class EcsClient(object):
    def __init__(self, config):
        # 区域设置
        self.region_id = config.get("region_id", "cn-beijing")
        self.ak_id = config.get("ak_id")
        self.ak_secret = config.get("ak_secret")
        self.config = config
        self.clt = client.AcsClient(self.ak_id, self.ak_secret, self.region_id)
        # pass

    # 创建请求
    @staticmethod
    def build_request(config):
        request = RunInstancesRequest()
        #  实例所属的地域区
        request.set_ZoneId(config.get("zone_id")) if config.get("zone_id") else None
        # 交换机ID
        request.set_VSwitchId(config.get("vswitch_id")) if config.get("vswitch_id") else None
        # 镜像ID
        request.set_ImageId(config.get("image_id")) if config.get("image_id") else None
        # 安全组ID
        request.set_SecurityGroupId(config.get("security_group_id")) if config.get("security_group_id") else None
        # 实例规格
        request.set_InstanceType(config.get("instance_type")) if config.get("instance_type") else None
        # 优化本地盘
        request.set_IoOptimized(config.get("io_optimized")) if config.get("io_optimized") else None
        # 高效云盘
        request.set_SystemDiskCategory(config.get("system_disk_category")) if config.get(
            "system_disk_category") else None
        # 默认40G
        request.set_SystemDiskSize(config.get("system_disk_size")) if config.get("system_disk_size") else None
        request.set_SpotStrategy("SpotAsPriceGo")  # 竞价模式自动出价
        request.set_PasswordInherit(True)  # 使用镜像用户名密码
        # 实例名称
        request.set_InstanceName(config.get("instance_name")) if config.get("instance_name") else "auto_create_instance"
        # 保护期
        request.set_SpotDuration(config.get("spot_duration")) if config.get("spot_duration") else 0
        return request

    # 创建实例并启动
    def create_multiple_instances(self, amount):
        '''
        创建实例
        :param amount: 创建数量
        :param config: 实例配置见 build_request
        :return:
        '''
        request = self.build_request(self.config)
        request.set_Amount(amount)
        return self._execute_request(request, amount)

    # 发送API请求
    def _send_request(self, request):
        request.set_accept_format('json')
        try:
            response_str = self.clt.do_action_with_exception(request)
            logging.info(response_str)
            response_detail = json.loads(response_str)
            return response_detail
        except Exception as e:
            logging.error(e)
            return None

    def _execute_request(self, request, amount):
        response = self._send_request(request)
        if response.get('Code') is None:
            instance_ids = response.get('InstanceIdSets').get('InstanceIdSet')
            instances = []
            while len(instances) < amount:
                time.sleep(10)
                instances = self.check_instance_running(instance_ids)
            return instances

    # 检查实例的运行状态
    def check_instance_running(self, instance_ids):
        request = DescribeInstancesRequest()
        request.set_InstanceIds(json.dumps(instance_ids))
        response = self._send_request(request)
        if response.get('Code') is None:
            instances_list = response.get('Instances').get('Instance')
            instances = []
            for instance_detail in instances_list:
                if instance_detail.get('Status') == "Running":
                    vpc_attributes = instance_detail.get('VpcAttributes', {}).get('PrivateIpAddress', {}).get(
                        'IpAddress',[])
                    intranet_ip = vpc_attributes[0] if len(vpc_attributes) > 0 else None
                    instances.append((instance_detail.get('InstanceId'), intranet_ip))
            return instances

    def release_instance(self, instance_id, force=False):
        """
        根据实例id删除实例
        :param instance_id: 实例id
        :param force: 当值为True时，运行的中实例也会被释放
        :return:
        """
        request = DeleteInstanceRequest()
        request.set_InstanceId(instance_id)
        request.set_Force(force)
        return self._send_request(request)

    # 检查实例的运行状态
    def check_instance_status(self, instance_ids):
        instances = []
        request = DescribeInstancesRequest()
        request.set_InstanceIds(json.dumps(instance_ids))
        response = self._send_request(request)
        if response.get('Code') is None:
            instances_list = response.get('Instances').get('Instance')
            for instance_detail in instances_list:
                vpc_attributes = instance_detail.get('VpcAttributes', {}).get('PrivateIpAddress', {}).get('IpAddress',
                                                                                                          [])
                intranet_ip = vpc_attributes[0] if len(vpc_attributes) > 0 else None
                instances.append((instance_detail.get('InstanceId'), intranet_ip))
            return instances
        else:
            return None

    def add_instance(self, num):
        for i in range(5):
            try:
                instance_ids = self.create_multiple_instances(num)
                if instance_ids:
                    return instance_ids
            except Exception as e:
                print(e)
                time.sleep(5)

    def delete_instance(self, instance_ids):
        delete_false = []
        for instance_id in instance_ids:
            try:
                result = self.release_instance(instance_id, force=True)
                if not result:
                    delete_false.append(instance_id)
            except Exception as e:
                print(e)
                time.sleep(2)
                delete_false.append(instance_id)
        return delete_false

    def instance_status(self, instance_ids):
        for r in range(5):
            try:
                result = self.check_instance_status(instance_ids)
                if result:
                    return result
            except Exception as e:
                print(e)
                time.sleep(2)


if __name__ == '__main__':
    config = {
        "ak_id": "LTAI4G5x9aoZx8dDamQ7vfZi",
        "region_id": "cn-beijing",
        "ak_secret": "Bk98FsbPYXcJe72n1bG3Ssf73acuNh",
        "image_id": "m-2zeihn2b3vxj7s5z1zxp",
        "instance_type": "ecs.g5.xlarge",
        "security_group_id": "sg-2ze6zhxrqy9vueae27xy",
        "vswitch_id": "vsw-2ze1n1k3mo3fv2irsfdps"
    }
    ecs = EcsClient(config)
    # result=ecs.create_multiple_instances(1)
    # print(result)

    # result = ecs.check_instance_status(["i-2ze4j0syekh1oq9017ok"])
    # print(result)
    # result = ecs.release_instance("i-2ze4j0syekh1oq9017ok", force=True)
    # print(result)