resource_tracker.cloud_info

Detect cloud environment (provider, region, instance type) via VM metadata services.

Functions:

Name	Description
get_cloud_info	Detect cloud environment and return standardized information on provider, region, and instance type.

get_cloud_info cached

get_cloud_info()

Detect cloud environment and return standardized information on provider, region, and instance type.

Returns:

Type	Description
dict	A dictionary containing standardized cloud information: vendor: The cloud provider (aws, gcp, azure, hcloud, upcloud, alicloud, ovh), or "unknown" instance_type: The instance type/size/flavor, or "unknown" region: The region/zone where the instance is running, or "unknown" discovery_time: The time taken to discover the cloud environment, in seconds

Source code in resource_tracker/cloud_info.py

@cache
def get_cloud_info() -> dict:
    """
    Detect cloud environment and return standardized information on provider, region, and instance type.

    Returns:
        A dictionary containing standardized cloud information:

            - `vendor`: The cloud provider (aws, gcp, azure, hcloud, upcloud, alicloud, ovh), or "unknown"
            - `instance_type`: The instance type/size/flavor, or "unknown"
            - `region`: The region/zone where the instance is running, or "unknown"
            - `discovery_time`: The time taken to discover the cloud environment, in seconds
    """
    start_time = time()
    check_functions = [
        _check_aws,
        _check_gcp,
        _check_azure,
        _check_hetzner,
        _check_upcloud,
        _check_alicloud,
        _check_ovh,
    ]

    # run checks in parallel, return early if any check succeeds
    with ThreadPoolExecutor(max_workers=len(check_functions)) as executor:
        futures = {executor.submit(check_fn): check_fn for check_fn in check_functions}
        pending = set(futures.keys())
        while pending:
            done, pending = wait(pending, return_when=FIRST_COMPLETED)
            for future in done:
                with suppress(Exception):
                    info = future.result()
                    if info:
                        # stop all remaining checks early
                        for f in pending:
                            f.cancel()
                        return info | {"discovery_time": time() - start_time}

    return {
        "vendor": "unknown",
        "instance_type": "unknown",
        "region": "unknown",
        "discovery_time": time() - start_time,
    }