_alicloud

sc_crawler.vendors._alicloud

Functions:

Name	Description
inventory_compliance_frameworks	Manual list of compliance frameworks known for Alibaba Cloud.
inventory_regions	List all available Alibaba Cloud regions.
inventory_zones	List all availability zones.
inventory_servers	List all server types at Alibaba Cloud using the DescribeInstanceTypes API endpoint.
inventory_server_prices	Fetch server pricing and regional availability using the QuerySkuPriceListRequest API endpoint.
inventory_storages	List all block storage offerings.
inventory_storage_prices	Fetch server prices using the QuerySkuPriceListRequest API endpoint.
inventory_traffic_prices	Collect inbound and outbound traffic prices of Alibaba Cloud regions.
inventory_ipv4_prices	Static IPv4 pricing of Alibaba Cloud regions.

inventory_compliance_frameworks

inventory_compliance_frameworks(vendor)

Manual list of compliance frameworks known for Alibaba Cloud.

Resources: https://www.alibabacloud.com/en/trust-center/compliance

Source code in sc_crawler/vendors/_alicloud.py

def inventory_compliance_frameworks(vendor):
    """Manual list of compliance frameworks known for Alibaba Cloud.

    Resources: <https://www.alibabacloud.com/en/trust-center/compliance>
    """
    return map_compliance_frameworks_to_vendor(
        vendor.vendor_id, ["hipaa", "soc2t2", "iso27001"]
    )

inventory_regions

inventory_regions(vendor)

List all available Alibaba Cloud regions.

Data sources:

• https://api.alibabacloud.com/document/Ecs/2014-05-26/DescribeRegions
• Foundation year collected from https://www.alibabacloud.com/en/global-locations?_p_lc=1
• Aliases (old region names) collected from https://help.aliyun.com/zh/user-center/product-overview/regional-name-change-announcement

Source code in sc_crawler/vendors/_alicloud.py

def inventory_regions(vendor):
    """
    List all available Alibaba Cloud regions.

    Data sources:

    - <https://api.alibabacloud.com/document/Ecs/2014-05-26/DescribeRegions>
    - Foundation year collected from <https://www.alibabacloud.com/en/global-locations?_p_lc=1>
    - Aliases (old region names) collected from <https://help.aliyun.com/zh/user-center/product-overview/regional-name-change-announcement>
    """
    request = DescribeRegionsRequest(accept_language="en-US")
    response = _ecs_client().describe_regions(request)
    regions = [region.to_map() for region in response.body.regions.region]

    items = []
    for region in regions:
        location = locations[region.get("RegionId")]
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.get("RegionId"),
                "name": region.get("LocalName"),
                "api_reference": region.get("RegionId"),
                "display_name": f"{location['city']} ({location['country_id']})",
                "aliases": location.get("alias", []),
                "country_id": location.get("country_id"),
                "state": None,  # not available
                "city": location.get("city"),
                "address_line": None,  # not available
                "zip_code": None,  # not available
                "lon": location.get("lon"),
                "lat": location.get("lat"),
                "founding_year": location.get("founding_year"),
                # "Clean electricity accounted for 56.0% of the total electricity consumption at Alibaba Cloud's self-built data centers"
                # https://www.alibabagroup.com/en-US/esg?spm=a3c0i.28208492.4078276800.1.3ee123b78lagGT
                "green_energy": None,
            }
        )
    return items

inventory_zones

inventory_zones(vendor)

List all availability zones.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_zones(vendor):
    """List all availability zones."""
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for zone(s)", total=len(vendor.regions)
    )
    clients = _ecs_clients(vendor)

    def fetch_zones_for_region(region):
        """Worker function to fetch zones for a single region."""
        request = DescribeZonesRequest(
            region_id=region.region_id, accept_language="en-US"
        )
        try:
            response = clients[region.region_id].describe_zones(request)
            zone_items = []
            for zone in response.body.to_map()["Zones"]["Zone"]:
                zone_items.append(
                    {
                        "vendor_id": vendor.vendor_id,
                        "region_id": region.region_id,
                        "zone_id": zone.get("ZoneId"),
                        "name": zone.get("LocalName"),
                        "api_reference": zone.get("ZoneId"),
                        "display_name": zone.get("LocalName"),
                    }
                )
            return zone_items
        except Exception as e:
            logger.error(f"Failed to get zones for region {region.region_id}: {e}")
            return []
        finally:
            vendor.progress_tracker.advance_task()

    with ThreadPoolExecutor(max_workers=8) as executor:
        items = executor.map(fetch_zones_for_region, vendor.regions)
    items = list(chain.from_iterable(items))
    vendor.progress_tracker.hide_task()
    return items

inventory_servers

inventory_servers(vendor)

List all server types at Alibaba Cloud using the DescribeInstanceTypes API endpoint.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_servers(vendor):
    """List all server types at Alibaba Cloud using the `DescribeInstanceTypes` API endpoint."""
    client = _ecs_client()
    request = DescribeInstanceTypesRequest(max_results=1000)
    response = client.describe_instance_types(request)
    instance_types = [
        instance_type.to_map()
        for instance_type in response.body.instance_types.instance_type
    ]
    while response.body.next_token:
        request = DescribeInstanceTypesRequest(
            max_results=1000, next_token=response.body.next_token
        )
        response = client.describe_instance_types(request)
        for instance_type in response.body.instance_types.instance_type:
            instance_types.append(instance_type.to_map())

    CPU_ARCH_MAP = {"X86": CpuArchitecture.X86_64, "ARM": CpuArchitecture.ARM64}
    STORAGE_CATEGORY_MAP = {
        "": None,
        "local_ssd_pro": StorageType.SSD,
        "local_hdd_pro": StorageType.HDD,
    }

    def drop_zero_value(x):
        return None if x == 0 else x

    items = []
    for instance_type in instance_types:
        family = instance_type.get("InstanceTypeFamily")
        vcpus = instance_type.get("CpuCoreCount")
        cpu_model = instance_type.get("PhysicalProcessorModel")
        memory_size_mb = int((instance_type.get("MemorySize") * 1024))
        memory_size_gb = (
            memory_size_mb // 1024
            if memory_size_mb >= 1024
            else round(memory_size_mb / 1024, 2)
        )
        storage_size = int(
            instance_type.get("LocalStorageAmount", 0)
            * instance_type.get("LocalStorageCapacity", 0)
            # convert GiB to GB
            * 1024**3
            / 1000**3
        )
        storage_type = STORAGE_CATEGORY_MAP[instance_type["LocalStorageCategory"]]
        gpu_count = instance_type.get("GPUAmount", 0)
        gpu_memory_per_gpu = instance_type.get("GPUMemorySize", 0) * 1024  # GiB -> MiB
        gpu_memory_total = gpu_count * gpu_memory_per_gpu
        gpu_model = _standardize_gpu_model(instance_type["GPUSpec"])
        description_parts = [
            f"{vcpus} vCPUs",
            f"{memory_size_gb} GiB RAM",
            f"{storage_size} GB {storage_type.value if storage_type else ''} storage",
            (
                f"{gpu_count}x{gpu_model} {gpu_memory_per_gpu} GiB VRAM"
                if gpu_count and gpu_model
                else None
            ),
        ]
        description = f"{family} family ({', '.join(filter(None, description_parts))})"

        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "server_id": instance_type.get("InstanceTypeId"),
                "name": instance_type.get("InstanceTypeId"),
                "api_reference": instance_type.get("InstanceTypeId"),
                "display_name": instance_type.get("InstanceTypeId"),
                "description": description,
                "family": family,
                "vcpus": vcpus,
                "hypervisor": "KVM",
                "cpu_allocation": _determine_cpu_allocation_type(instance_type),
                "cpu_cores": instance_type.get("CpuCoreCount", 0),
                "cpu_speed": drop_zero_value(instance_type.get("CpuSpeedFrequency")),
                "cpu_architecture": CPU_ARCH_MAP[instance_type.get("CpuArchitecture")],
                "cpu_manufacturer": _extract_manufacturer(cpu_model),
                "cpu_family": _extract_family(cpu_model),
                "cpu_model": _standardize_cpu_model(cpu_model),
                "cpu_l1_cache": None,
                "cpu_l2_cache": None,
                "cpu_l3_cache": None,
                "cpu_flags": [],
                "cpus": [],
                "memory_amount": memory_size_mb,
                "memory_generation": None,
                "memory_speed": None,
                "memory_ecc": None,
                "gpu_count": gpu_count,
                "gpu_memory_min": drop_zero_value(int(gpu_memory_per_gpu)),
                "gpu_memory_total": drop_zero_value(int(gpu_memory_total)),
                # TODO fill in from GPUSpec? or just let the inspector fill it in?
                "gpu_manufacturer": None,
                "gpu_family": None,
                "gpu_model": gpu_model,
                "gpus": [],
                "storage_size": storage_size,
                "storage_type": storage_type,
                "storages": [],
                "network_speed": drop_zero_value(
                    instance_type.get("InstanceBandwidthRx", 0) / 1024 / 1000
                ),
                "inbound_traffic": 0,
                "outbound_traffic": 0,
                "ipv4": 0,
            }
        )
    return items

inventory_server_prices

inventory_server_prices(vendor)

Fetch server pricing and regional availability using the QuerySkuPriceListRequest API endpoint.

Alternative approach could be looking at https://g.alicdn.com/aliyun/ecs-price-info-intl/2.0.375/price/download/instancePrice.json.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_server_prices(vendor):
    """Fetch server pricing and regional availability using the `QuerySkuPriceListRequest` API endpoint.

    Alternative approach could be looking at <https://g.alicdn.com/aliyun/ecs-price-info-intl/2.0.375/price/download/instancePrice.json>.
    """
    skus = _get_sku_prices(
        sku_type="server",
        extra_request_params={
            "price_entity_code": "instance_type",
            # filter for Linux prices only for now
            "price_factor_condition_map": {"vm_os_kind": ["linux"]},
        },
        vendor=vendor,
    )

    items = []
    unsupported_regions = set()
    region_availability_info: dict[str, list[dict]] = _get_resource_availability_info(
        vendor
    )

    for sku in skus:
        sku_region_id = sku["SkuFactorMap"]["vm_region_no"]
        region = get_region_by_id(sku_region_id, vendor)
        if not region:
            unsupported_regions.add(sku_region_id)
            continue
        for zone in region.zones:
            server_id = sku.get("SkuFactorMap", {}).get("instance_type")
            status = Status.INACTIVE
            zone_availability_info = next(
                (
                    r
                    for r in region_availability_info[region.region_id]
                    if r["ZoneId"] == zone.zone_id
                ),
                None,
            )
            if zone_availability_info:
                available_resource: list[dict] = zone_availability_info.get(
                    "AvailableResources", {}
                ).get("AvailableResource", [])
                supported_resource: list[dict] = (
                    next(
                        (r for r in available_resource if r["Type"] == "InstanceType"),
                        {},
                    )
                    .get("SupportedResources", {})
                    .get("SupportedResource", [])
                )
                server_info = next(
                    (r for r in supported_resource if r["Value"] == server_id),
                    {},
                )
                if server_info.get("StatusCategory") == "WithStock":
                    status = Status.ACTIVE

            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region.region_id,
                    "zone_id": zone.zone_id,
                    "server_id": server_id,
                    "operating_system": sku.get("SkuFactorMap").get("vm_os_kind"),
                    "allocation": Allocation.ONDEMAND,
                    "unit": PriceUnit.HOUR,
                    "price": sku.get("CskuPriceList")[0].get("Price"),
                    "price_upfront": 0,
                    "price_tiered": [],
                    "currency": sku.get("CskuPriceList")[0].get("Currency"),
                    "status": status,
                }
            )
    for unsupported_region in unsupported_regions:
        vendor.log(f"Found non-supported region: {unsupported_region}", level=WARN)
    return items

inventory_storages

inventory_storages(vendor)

List all block storage offerings.

Data sources:

• https://www.alibabacloud.com/help/en/ecs/user-guide/essds
• https://www.alibabacloud.com/help/en/ecs/developer-reference/api-ecs-2014-05-26-createdisk

Source code in sc_crawler/vendors/_alicloud.py

def inventory_storages(vendor):
    """List all block storage offerings.

    Data sources:

    - <https://www.alibabacloud.com/help/en/ecs/user-guide/essds>
    - <https://www.alibabacloud.com/help/en/ecs/developer-reference/api-ecs-2014-05-26-createdisk>
    """
    disk_info = [
        # NOTE there's only a single `cloud_essd` ID at Alibaba Cloud,
        # but we suffix with the performance level (PL0, PL1, PL2, PL3)
        # to differentiate them as these are products with very different characteristics
        {
            "name": "cloud_essd-pl0",
            "min_size": 1,
            "max_size": 65536,
            "max_iops": 10000,
            "max_tp": 1440,
            "info": "Enterprise SSD with performance level 0.",
        },
        {
            "name": "cloud_essd-pl1",
            "min_size": 20,
            "max_size": 65536,
            "max_iops": 50000,
            "max_tp": 2800,
            "info": "Enterprise SSD with performance level 1.",
        },
        {
            "name": "cloud_essd-pl2",
            "min_size": 461,
            "max_size": 65536,
            "max_iops": 100000,
            "max_tp": 6000,
            "info": "Enterprise SSD with performance level 2.",
        },
        {
            "name": "cloud_essd-pl3",
            "min_size": 1261,
            "max_size": 65536,
            "max_iops": 1000000,
            "max_tp": 32000,
            "info": "Enterprise SSD with performance level 3.",
        },
        {
            "name": "cloud_ssd",
            "min_size": 20,
            "max_size": 32768,
            "max_iops": 20000,
            "max_tp": 256,
            "info": "Standard SSD.",
        },
        {
            "name": "cloud_efficiency",
            "min_size": 20,
            "max_size": 32768,
            "max_iops": 3000,
            "max_tp": 80,
            "info": "Ultra Disk, older generation.",
        },
        {
            "name": "cloud",
            "min_size": 5,
            "max_size": 2000,
            "max_iops": 300,
            "max_tp": 40,
            "info": "Lowest cost HDD.",
        },
    ]

    items = []
    for disk in disk_info:
        items.append(
            {
                "storage_id": disk.get("name"),
                "vendor_id": vendor.vendor_id,
                "name": disk.get("name"),
                "description": disk.get("info"),
                "storage_type": (
                    StorageType.HDD if disk.get("name") == "cloud" else StorageType.SSD
                ),
                "max_iops": disk.get("max_iops"),
                "max_throughput": disk.get("max_tp"),
                "min_size": disk.get("min_size"),
                "max_size": disk.get("max_size"),
            }
        )
    return items

inventory_storage_prices

inventory_storage_prices(vendor)

Fetch server prices using the QuerySkuPriceListRequest API endpoint.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_storage_prices(vendor):
    """Fetch server prices using the `QuerySkuPriceListRequest` API endpoint."""
    skus = _get_sku_prices(
        sku_type="storage",
        extra_request_params={"price_entity_code": "datadisk"},
        vendor=vendor,
    )

    items = []
    unsupported_regions = set()
    for sku in skus:
        storage_id = sku["SkuFactorMap"]["datadisk_category"]
        pl = sku["SkuFactorMap"]["datadisk_performance_level"]
        if storage_id in ["cloud", "cloud_ssd", "cloud_efficiency"]:
            # no diff in performance levels, pick one
            if pl != "PL1":
                continue
        else:
            # keep the 4 performance levels
            if pl not in ["PL0", "PL1", "PL2", "PL3"]:
                continue
            storage_id = storage_id + "-" + pl.lower()
        region_id = sku["SkuFactorMap"]["vm_region_no"]
        region = get_region_by_id(region_id, vendor)
        if not region:
            unsupported_regions.add(region_id)
            continue
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.region_id,
                "storage_id": storage_id,
                "unit": PriceUnit.GB_MONTH,
                "price": sku["CskuPriceList"][0]["Price"],
                "currency": sku["CskuPriceList"][0]["Currency"],
            }
        )
    for unsupported_region in unsupported_regions:
        vendor.log(f"Found non-supported region: {unsupported_region}", level=WARN)
    return items

inventory_traffic_prices

inventory_traffic_prices(vendor)

Collect inbound and outbound traffic prices of Alibaba Cloud regions.

Inbound is free as per https://www.alibabacloud.com/help/en/ecs/public-bandwidth. Outbound traffic pricing collected from the QuerySkuPriceListRequest API endpoint.

Account level tiering information can be found at https://www.alibabacloud.com/help/en/cdt/internet-data-transfers/#4a98c9ee8eemn.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_traffic_prices(vendor):
    """Collect inbound and outbound traffic prices of Alibaba Cloud regions.

    Inbound is free as per <https://www.alibabacloud.com/help/en/ecs/public-bandwidth>.
    Outbound traffic pricing collected from the `QuerySkuPriceListRequest` API endpoint.

    Account level tiering information can be found at <https://www.alibabacloud.com/help/en/cdt/internet-data-transfers/#4a98c9ee8eemn>.
    """
    items = []
    skus = _get_sku_prices(
        sku_type="traffic",
        extra_request_params={"price_entity_code": "vm_flow_out"},
        # vendor=vendor,
    )
    unsupported_regions = set()
    for sku in skus:
        region_id = sku["SkuFactorMap"]["vm_region_no"]
        region = get_region_by_id(region_id, vendor)
        if not region:
            unsupported_regions.add(region_id)
            continue
        price = next(p for p in sku["CskuPriceList"] if float(p["Price"]) > 0)
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.region_id,
                "price": price["Price"],
                "price_tiered": [],
                "currency": price["Currency"],
                "unit": PriceUnit.GB_MONTH,
                "direction": TrafficDirection.OUT,
            }
        )
        # incoming traffic is free
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.region_id,
                "price": 0,
                "price_tiered": [],
                "currency": price["Currency"],
                "unit": PriceUnit.GB_MONTH,
                "direction": TrafficDirection.IN,
            }
        )
    for unsupported_region in unsupported_regions:
        vendor.log(f"Found non-supported region: {unsupported_region}", level=WARN)
    return items

inventory_ipv4_prices

inventory_ipv4_prices(vendor)

Static IPv4 pricing of Alibaba Cloud regions.

Static (not Elastic) IP addresses are free, you only pay for bandwidth or traffic as per https://www.alibabacloud.com/help/en/ecs/user-guide/public-ip-address?spm=a2c63.p38356.0.i1#52c0fa8bbcee6.

Source code in sc_crawler/vendors/_alicloud.py

def inventory_ipv4_prices(vendor):
    """Static IPv4 pricing of Alibaba Cloud regions.

    Static (not Elastic) IP addresses are free, you only pay for bandwidth or traffic
    as per <https://www.alibabacloud.com/help/en/ecs/user-guide/public-ip-address?spm=a2c63.p38356.0.i1#52c0fa8bbcee6>.
    """
    items = []
    for region in vendor.regions:
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.region_id,
                "price": 0,
                "currency": "USD",
                "unit": PriceUnit.MONTH,
            }
        )
    return items