c3mc-ec2create-describe-instance-type-offerings

#!/usr/bin/env /data/Software/mydan/python3/bin/python3
# -*- coding: utf-8 -*-

import sys
import json
import subprocess
import concurrent.futures
import threading

import boto3


def fetch_instance_types(client, parts, data, lock):
    instance_types = client.describe_instance_types(InstanceTypes=parts)
    if instance_types["InstanceTypes"]:
        with lock:
            data.extend(instance_types["InstanceTypes"])


def filter_by_architecture(client, architecture, instance_type_list):
    """
    根据架构类型过滤实例类型列表
    """
    data = []
    step_length = 100
    max_workers = 100  # 最大并发数
    lock = threading.Lock()

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for i in range(0, len(instance_type_list), step_length):
            parts = instance_type_list[i:i+step_length]
            futures.append(executor.submit(fetch_instance_types, client, parts, data, lock))
        
        # 等待所有任务完成
        concurrent.futures.wait(futures)

    result = {}
    for instance_type_info in data:
        if not instance_type_info['ProcessorInfo'] or not instance_type_info['ProcessorInfo']['SupportedArchitectures']:
            continue

        if architecture in instance_type_info['ProcessorInfo']['SupportedArchitectures']:
            result[instance_type_info['InstanceType']] = {
                "instance_type": instance_type_info['InstanceType'],
                "cpu": instance_type_info["VCpuInfo"]["DefaultVCpus"],
                "memory": int(instance_type_info["MemoryInfo"]["SizeInMiB"] / 1024),
            }
    return list(result.values())


def describe_instance_type_offerings(access_id, access_key, region, availability_zone, architecture):
    """
        查询ec2的实例类型列表
    """
    client = boto3.client(
        "ec2",
        aws_access_key_id=access_id,
        aws_secret_access_key=access_key,
        region_name=region
    )

    page_size = 1000

    data = []
    response = client.describe_instance_type_offerings(
        LocationType="availability-zone",
        Filters=[
            {
                'Name': 'location',
                'Values': [availability_zone]
            },
        ],
        MaxResults=page_size
    )
    data.extend(response["InstanceTypeOfferings"])
    next_token = response["NextToken"] if "NextToken" in response else ""
    while next_token != "":
        response = client.describe_instance_type_offerings(
            LocationType="availability-zone",
            Filters=[
                {
                    'Name': 'location',
                    'Values': [availability_zone]
                },
            ],
            NextToken=next_token,
            MaxResults=page_size
        )
        data.extend(response["InstanceTypeOfferings"])

        next_token = response["NextToken"] if "NextToken" in response else ""
    instance_type_list = [item["InstanceType"] for item in data]
    return filter_by_architecture(client, architecture, instance_type_list)


def sort_instance_type_list(item):
    parts = item["instance_type"].split(".")
    instance_type_sort_key = parts[0]
    return (instance_type_sort_key, item["cpu"], item["memory"])


def main(params):
    account = params["account"]
    region = params["region"]
    availability_zone = params["az"]
    architecture = params["architecture"]

    key_info = json.loads(
        subprocess.getoutput(f"c3mc-get-account-info aws {account}")
    )

    instance_type_list = describe_instance_type_offerings(
        key_info["access_id"], key_info["access_key"], region, availability_zone, architecture)

    instance_type_list = sorted(instance_type_list, key=sort_instance_type_list)
    for instance_type in instance_type_list:
        print(json.dumps(instance_type, default=str))


if __name__ == '__main__':
    l = list(sys.stdin)
    if not l or len(l) > 1:
        raise type('WrongInputData', (Exception,), {})('数据格式不对, 需要一行json字符串"')

    main(json.loads(l[0]))