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), or "unknown"instance_type: The instance type/size/flavor, or "unknown"region: The region/zone where the instance is running, or "unknown"
|
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), or "unknown"
- `instance_type`: The instance type/size/flavor, or "unknown"
- `region`: The region/zone where the instance is running, or "unknown"
"""
start_time = time()
check_functions = [
_check_aws,
_check_gcp,
_check_azure,
_check_hetzner,
_check_upcloud,
]
# 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,
}
|