contrib/cloud/ali-import - master - githubmirror/ipxe

contrib/cloud/ali-import
713 строк28 KB
Michael Brown
[cloud] Separate snapshot deletion from image deletion for Alibaba Cloud
24 апр 2026, 14:55
24 апр 2026, 14:55a42daf6
#!/usr/bin/env python3
#
# Doing anything in Alibaba Cloud is unnecessarily difficult and
# tedious due to a combination of poor and inconsistent API design,
# high API call failure rates, and Chinese state censorship laws.
#
# We resort to a mixture of strategies to get images imported to all
# regions:
#
#   - For regions with working OSS that are not blocked by Chinese
#     state censorship laws, upload the image files to an OSS bucket
#     and then import the images.
#
#   - For regions with working OSS that are blocked by Chinese state
#     censorship laws but that have working FC, use a temporary FC
#     function to copy the image files from the uncensored OSS buckets
#     and then import the images.  Attempt downloads from a variety of
#     uncensored buckets, since cross-region OSS traffic tends to
#     experience a failure rate of around 10% of requests.
#
#   - For regions that have working OSS but are blocked by Chinese
#     state censorship laws and do not have working FC, or for regions
#     that don't even have working OSS, resort to using CopyImage to
#     copy the previously imported images from another region.  Spread
#     the imports across as many source regions as possible to
#     minimise the effect of the CopyImage rate limiting.

import argparse
import base64
from collections import namedtuple
from concurrent.futures import ThreadPoolExecutor, as_completed
import datetime
import http
import io
from itertools import cycle, groupby
import json
import logging
from operator import itemgetter
from pathlib import Path
import random
import subprocess
import tempfile
import time
from uuid import uuid4
import zipfile

import alibabacloud_credentials as credentials
import alibabacloud_credentials.client
import alibabacloud_credentials.models
import alibabacloud_ecs20140526 as ecs
import alibabacloud_ecs20140526.client
import alibabacloud_ecs20140526.models
import alibabacloud_fc20230330 as fc
import alibabacloud_fc20230330.client
import alibabacloud_fc20230330.models
import alibabacloud_oss_v2 as oss
import alibabacloud_ram20150501 as ram
import alibabacloud_ram20150501.client
import alibabacloud_ram20150501.models
import alibabacloud_sts20150401 as sts
import alibabacloud_sts20150401.client
import alibabacloud_sts20150401.models
import alibabacloud_tea_openapi as openapi
import alibabacloud_tea_openapi.client
import alibabacloud_tea_openapi.models
import alibabacloud_tea_util as util
import alibabacloud_tea_util.client
import alibabacloud_tea_util.models

logger = logging.getLogger('ali-import')

ECS_ENDPOINT = 'ecs.aliyuncs.com'
RAM_ENDPOINT = 'ram.aliyuncs.com'
STS_ENDPOINT = 'sts.aliyuncs.com'

FC_NODE_RUNTIME = 'nodejs20'
FC_TIMEOUT_SEC = 120
FC_MEMORY_SIZE_MB = 128
FC_SOURCE_COUNT = 10

OSS_FORBIDDEN_REGION_CODE = 'ForbidCreateNewBucket'
OSS_BUCKET_NAME_LEN = 63

IPXE_STORAGE_PREFIX = 'ipxe-upload-temp-'
IPXE_SNAPSHOT_DELETE_TAG = 'ipxe-snapshot-delete'

POLL_INTERVAL_SEC = 5
POLL_MAX_RETRIES = 100

# Experimentation suggests Alibaba Cloud API calls are extremely
# unreliable, with a failure rate around 1%.  It is therefore
# necessary to allow for retrying basically every API call.
#
# Some API calls (e.g. DescribeImages or ModifyImageAttribute) are
# naturally idempotent and so safe to retry.  Some non-idempotent API
# calls (e.g. CopyImage) support explicit idempotence tokens.  The
# remaining API calls may simply fail on a retry, if the original
# request happened to succeed but failed to return a response.
#
# We could write convoluted retry logic around the non-idempotent
# calls, but this would substantially increase the complexity of the
# already unnecessarily complex code.  For now, we assume that
# retrying non-idempotent requests is probably more likely to fix
# transient failures than to cause additional problems.
#
RUNTIME_OPTS = util.models.RuntimeOptions(
    autoretry=True,
    max_attempts=5,
    connect_timeout=10000,
    read_timeout=120000,
)

# For regions in mainland China, the Chinese state censorship laws
# prohibit direct access to OSS bucket contents.
#
# We work around this restriction by creating a temporary Function
# Compute function in each region to access OSS via the internal OSS
# endpoints, which are not subject to these restrictions.  Yes, this
# is somewhat absurd.
#
IPXE_CENSORSHIP_BYPASS_FUNCTION = f'''
const prefix = "{IPXE_STORAGE_PREFIX}";
''' + '''
const assert = require("node:assert");
const OSS = require("ali-oss");
exports.handler = async (event, context) => {
  const payload = JSON.parse(event.toString());
  console.log(JSON.stringify(payload));
  const sources = payload.sources || {};
  const dest = new OSS({
    region: "oss-" + context.region,
    internal: true,
    bucket: payload.bucket,
    accessKeyId: context.credentials.accessKeyId,
    accessKeySecret: context.credentials.accessKeySecret,
    stsToken: context.credentials.securityToken,
  });
  const current = ((await dest.listV2({prefix: prefix})).objects || [])
    .map(x => x.name);
  const wanted = Object.keys(sources);
  const add = wanted.filter(x => ! current.includes(x));
  const del = current.filter(x => ! wanted.includes(x));
  assert(add.every(x => x.startsWith(prefix)));
  assert(del.every(x => x.startsWith(prefix)));
  if (add.length)
    console.log("Creating: " + add.sort().join(", "));
  if (del.length)
    console.log("Deleting: " + del.sort().join(", "));
  const copy = async (key) => {
    for (const url of sources[key]) {
      console.log("Downloading " + key + " from " + url);
      try {
        const download = await fetch(url, {signal: AbortSignal.timeout(15000)});
        if (! download.ok)
          throw new Error(download.status);
        const content = await download.arrayBuffer();
        console.log("Downloaded " + key);
        console.log("Uploading " + key);
        await dest.put(key, Buffer.from(content));
        console.log("Uploaded " + key);
        return;
      } catch (err) {
        console.error("Download failed", err);
      }
    }
    throw new Error("All downloads failed for " + key);
  };
  await Promise.all([
    ...add.map(copy),
    ...(del.length ? [dest.deleteMulti(del)] : []),
  ]);
  console.log("Finished");
};
'''

Clients = namedtuple('Clients', ['region', 'censored', 'ecs', 'fc', 'oss'])
Image = namedtuple('Image',
                   ['path', 'family', 'name', 'arch', 'mode', 'key', 'public'])

def image(filename, basefamily, basename, public):
    """Construct image description"""
    with tempfile.NamedTemporaryFile(mode='w+t') as mtoolsrc:
        mtoolsrc.writelines([
            'drive D:', f'file="{filename}"',
            'drive P:', f'file="{filename}"', 'partition=4',
        ])
        mtoolsrc.flush()
        mdir = subprocess.run(['mdir', '-b', 'D:/EFI/BOOT', 'P:/EFI/BOOT'],
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              check=False, env={'MTOOLSRC': mtoolsrc.name})
    mapping = {
        b'BOOTX64.EFI': 'x86_64',
        b'BOOTAA64.EFI': 'arm64',
    }
    uefi = [v for k, v in mapping.items() if k in mdir.stdout]
    suffix = ('-uefi-%s' % uefi[0].replace('_', '-') if len(uefi) == 1 else
              '-uefi-multi' if uefi else '')
    path = Path(filename)
    family = '%s%s' % (basefamily, suffix)
    name = '%s%s' % (basename, suffix)
    arch = uefi[0] if len(uefi) == 1 else None if uefi else 'x86_64'
    mode = 'UEFI' if uefi else 'BIOS'
    key = '%s%s' % (IPXE_STORAGE_PREFIX, uuid4())
    return Image(path, family, name, arch, mode, key, public)

def all_regions():
    """Get list of all regions"""
    cred = credentials.client.Client()
    conf = openapi.models.Config(credential=cred, endpoint=ECS_ENDPOINT)
    client = ecs.client.Client(conf)
    req = ecs.models.DescribeRegionsRequest()
    rsp = client.describe_regions(req)
    regions = sorted(x.region_id for x in rsp.body.regions.region)
    return regions

def account_id():
    """Get account ID"""
    cred = credentials.client.Client()
    conf = openapi.models.Config(credential=cred, endpoint=STS_ENDPOINT)
    client = sts.client.Client(conf)
    rsp = client.get_caller_identity()
    return rsp.body.account_id

def role_arn(name):
    """Get role resource name"""
    cred = credentials.client.Client()
    conf = openapi.models.Config(credential=cred, endpoint=RAM_ENDPOINT)
    client = ram.client.Client(conf)
    req = ram.models.GetRoleRequest(role_name=name)
    rsp = client.get_role(req)
    return rsp.body.role.arn

def all_clients(region, account):
    """Construct all per-region clients"""
    cred = credentials.client.Client()
    ecsconf = openapi.models.Config(credential=cred, region_id=region)
    fcep = '%s.%s.fc.aliyuncs.com' % (account, region)
    fcconf = openapi.models.Config(credential=cred, endpoint=fcep)
    osscred = oss.credentials.EnvironmentVariableCredentialsProvider()
    ossconf = oss.config.Config(credentials_provider=osscred, region=region)
    clients = Clients(
        region=region,
        censored=region.startswith('cn-'),
        ecs=ecs.client.Client(ecsconf),
        fc=fc.client.Client(fcconf),
        oss=oss.client.Client(ossconf),
    )
    return clients

def delete_temp_function(clients, func):
    """Remove temporary function"""
    logger.info("delete function %s %s" % (clients.region, func))
    assert func.startswith(IPXE_STORAGE_PREFIX)
    clients.fc.delete_function_with_options(func, {}, RUNTIME_OPTS)

def create_temp_function(clients, role):
    """Create temporary function (and remove any stale temporary functions)"""
    req = fc.models.ListFunctionsRequest(prefix=IPXE_STORAGE_PREFIX)
    try:
        rsp = clients.fc.list_functions_with_options(req, {}, RUNTIME_OPTS)
    except openapi.client.UnretryableException:
        # AliCloud provides no other way to detect non-working regions
        return None
    funcs = [x.function_name for x in rsp.body.functions or ()]
    for func in funcs:
        delete_temp_function(clients, func)
    if not clients.censored:
        # Functions are not required in uncensored regions
        return None
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w') as zfh:
        zfh.writestr('index.js', IPXE_CENSORSHIP_BYPASS_FUNCTION)
    zf = base64.b64encode(buf.getvalue()).decode()
    code = fc.models.InputCodeLocation(zip_file=zf)
    func = '%s%s' % (IPXE_STORAGE_PREFIX, uuid4())
    body = fc.models.CreateFunctionInput(
        code=code,
        function_name=func,
        handler='index.handler',
        memory_size=FC_MEMORY_SIZE_MB,
        role=role,
        runtime=FC_NODE_RUNTIME,
        timeout=FC_TIMEOUT_SEC,
    )
    req = fc.models.CreateFunctionRequest(body=body)
    rsp = clients.fc.create_function_with_options(req, {}, RUNTIME_OPTS)
    logger.info("create function %s %s" % (clients.region, func))
    return func

def call_temp_function(clients, func, payload):
    """Call temporary function"""
    hdr = fc.models.InvokeFunctionHeaders(
        x_fc_invocation_type='Sync',
        x_fc_log_type='Tail',
    )
    body = json.dumps(payload)
    req = fc.models.InvokeFunctionRequest(body=body)
    rsp = clients.fc.invoke_function_with_options(func, req, hdr, RUNTIME_OPTS)
    log = base64.b64decode(rsp.headers.get('x-fc-log-result', b'')).decode()
    if rsp.status_code != http.HTTPStatus.OK:
        raise RuntimeError(rsp)
    if 'x-fc-error-type' in rsp.headers:
        raise RuntimeError(log)

def delete_temp_bucket(clients, func, bucket):
    """Remove temporary bucket"""
    logger.info("delete bucket %s %s" % (clients.region, bucket))
    assert bucket.startswith(IPXE_STORAGE_PREFIX)
    # Delete bucket contents
    if not clients.censored:
        # Uncensored region: use OSS API calls to delete contents
        req = oss.models.ListObjectsV2Request(
            bucket=bucket,
            prefix=IPXE_STORAGE_PREFIX,
        )
        rsp = clients.oss.list_objects_v2(req)
        delete = [x.key for x in rsp.contents or ()]
        if delete:
            req = oss.models.DeleteMultipleObjectsRequest(
                bucket=bucket,
                objects=[oss.models.DeleteObject(x) for x in delete],
            )
            rsp = clients.oss.delete_multiple_objects(req)
    elif func:
        # Censored region with FC: use function to delete contents
        payload = {'bucket': bucket}
        call_temp_function(clients, func, payload)
    else:
        # Censored region without FC: assume bucket must be empty,
        # since we could not have uploaded to it in the first place
        pass
    # Delete the now-empty bucket
    req = oss.models.DeleteBucketRequest(bucket=bucket)
    rsp = clients.oss.delete_bucket(req)

def create_temp_bucket(clients, func):
    """Create temporary bucket (and remove any stale temporary buckets)"""
    prefix = '%s%s-' % (IPXE_STORAGE_PREFIX, clients.region)
    req = oss.models.ListBucketsRequest(prefix=prefix)
    rsp = clients.oss.list_buckets(req)
    buckets = [x.name for x in rsp.buckets or ()]
    for bucket in buckets:
        delete_temp_bucket(clients, func, bucket)
    if clients.censored and not func:
        # We cannot use OSS in censored regions with no Function Compute
        return None
    bucket = ('%s%s' % (prefix, uuid4()))[:OSS_BUCKET_NAME_LEN]
    req = oss.models.PutBucketRequest(bucket=bucket)
    try:
        rsp = clients.oss.put_bucket(req)
    except oss.exceptions.OperationError as exc:
        # AliCloud provides no other way to detect non-working regions
        if exc.unwrap().code == OSS_FORBIDDEN_REGION_CODE:
            return None
        raise exc
    logger.info("create bucket %s %s" % (clients.region, bucket))
    return bucket

def upload_object(clients, bucket, image):
    """Upload disk image object to uncensored bucket"""
    logger.info("upload %s %s" % (clients.region, image.name))
    req = oss.models.PutObjectRequest(bucket=bucket, key=image.key)
    rsp = clients.oss.put_object_from_file(req, image.path)
    req = oss.models.GetObjectRequest(bucket=bucket, key=image.key)
    rsp = clients.oss.presign(req)
    return rsp.url

def copy_objects(clients, bucket, func, uploads):
    """Copy disk image objects to censored bucket from uncensored bucket"""
    logger.info("upload %s (censored)" % clients.region)
    payload = {
        'bucket': bucket,
        'sources': {
            key: random.choices([url for key, url in urls], k=FC_SOURCE_COUNT)
            for key, urls in groupby(sorted(
                ((image.key, url) for (region, image), url in uploads.items())
            ), key=itemgetter(0))
        }
    }
    call_temp_function(clients, func, payload)

def delete_image(clients, name):
    """Remove existing image (if applicable)"""
    req = ecs.models.DescribeImagesRequest(
        region_id=clients.region,
        image_name=name,
        image_owner_alias='self',
    )
    rsp = clients.ecs.describe_images_with_options(req, RUNTIME_OPTS)
    for image in rsp.body.images.image or ():
        logger.info("delete image %s %s (%s)" %
                    (clients.region, image.image_name, image.image_id))
        # Unpublish image
        if image.is_public:
            req = ecs.models.ModifyImageSharePermissionRequest(
                region_id=clients.region,
                image_id=image.image_id,
                is_public=False,
            )
            rsp = clients.ecs.modify_image_share_permission_with_options(
                req, RUNTIME_OPTS
            )
        # Tag associated snapshots for deletion
        for disk in image.disk_device_mappings.disk_device_mapping or ():
            snapshot_id = disk.snapshot_id
            tag = ecs.models.TagResourcesRequestTag(
                key=IPXE_SNAPSHOT_DELETE_TAG,
                value=IPXE_SNAPSHOT_DELETE_TAG,
            )
            req = ecs.models.TagResourcesRequest(
                region_id=clients.region,
                resource_type='snapshot',
                resource_id=[snapshot_id],
                tag=[tag],
            )
            rsp = clients.ecs.tag_resources_with_options(req, RUNTIME_OPTS)
        # Delete image
        req = ecs.models.DeleteImageRequest(
            region_id=clients.region,
            image_id=image.image_id
        )
        rsp = clients.ecs.delete_image_with_options(req, RUNTIME_OPTS)

def delete_snapshots(clients):
    """Remove stale snapshots left behind by deleted images"""
    tag = ecs.models.ListTagResourcesRequestTag(
        key=IPXE_SNAPSHOT_DELETE_TAG,
        value=IPXE_SNAPSHOT_DELETE_TAG,
    )
    req = ecs.models.ListTagResourcesRequest(
        region_id=clients.region,
        resource_type='snapshot',
        tag=[tag],
    )
    rsp = clients.ecs.list_tag_resources_with_options(req, RUNTIME_OPTS)
    for snapshot in rsp.body.tag_resources.tag_resource or ():
        logger.info("delete snapshot %s %s" %
                    (clients.region, snapshot.resource_id))
        req = ecs.models.DeleteSnapshotRequest(
            snapshot_id=snapshot.resource_id,
            force=True,
        )
        rsp = clients.ecs.delete_snapshot_with_options(req, RUNTIME_OPTS)

def wait_for_task(clients, task_id):
    """Wait for task to complete"""
    status = 'Unknowable'
    for i in range(POLL_MAX_RETRIES):
        time.sleep(POLL_INTERVAL_SEC)
        req = ecs.models.DescribeTasksRequest(
            region_id=clients.region,
            task_ids=task_id,
        )
        try:
            rsp = clients.ecs.describe_tasks_with_options(req, RUNTIME_OPTS)
        except openapi.client.UnretryableException:
            continue
        assert len(rsp.body.task_set.task) == 1
        assert rsp.body.task_set.task[0].task_id == task_id
        status = rsp.body.task_set.task[0].task_status
        if status not in ('Waiting', 'Processing'):
            break
    if status != 'Finished':
        raise RuntimeError(status)

def wait_for_image(clients, image_id):
    """Wait for image to become available"""
    status = 'Unknowable'
    for i in range(POLL_MAX_RETRIES):
        time.sleep(POLL_INTERVAL_SEC)
        req = ecs.models.DescribeImagesRequest(
            region_id=clients.region,
            image_id=image_id,
        )
        try:
            rsp = clients.ecs.describe_images_with_options(req, RUNTIME_OPTS)
        except openapi.client.UnretryableException:
            continue
        if len(rsp.body.images.image):
            assert len(rsp.body.images.image) == 1
            assert rsp.body.images.image[0].image_id == image_id
            status = rsp.body.images.image[0].status
            if status != 'Creating':
                break
    if status != 'Available':
        raise RuntimeError(status)

def import_image(clients, image, bucket):
    """Import image"""
    logger.info("import %s %s" % (clients.region, image.name))
    disk = ecs.models.ImportImageRequestDiskDeviceMapping(
        disk_image_size = 1,
        format = 'RAW',
        ossbucket = bucket,
        ossobject = image.key,
    )
    features = ecs.models.ImportImageRequestFeatures(
        imds_support='v1',
        nvme_support='supported',
    )
    req = ecs.models.ImportImageRequest(
        region_id=clients.region,
        image_name=image.name,
        architecture=image.arch,
        boot_mode=image.mode,
        disk_device_mapping=[disk],
        features=features,
        client_token=str(uuid4()),
    )
    rsp = clients.ecs.import_image_with_options(req, RUNTIME_OPTS)
    image_id = rsp.body.image_id
    task_id = rsp.body.task_id
    wait_for_task(clients, task_id)
    wait_for_image(clients, image_id)
    logger.info("image %s %s (%s)" %
                (clients.region, image.name, image_id))
    return image_id

def copy_image(clients, image, image_id, censored):
    """Copy imported image to censored region"""
    logger.info("import %s %s via %s" %
                (censored.region, image.name, clients.region))
    req = ecs.models.CopyImageRequest(
        region_id=clients.region,
        image_id=image_id,
        destination_region_id=censored.region,
        destination_image_name=image.name,
        client_token=str(uuid4()),
    )
    rsp = clients.ecs.copy_image_with_options(req, RUNTIME_OPTS)
    copy_id = rsp.body.image_id
    wait_for_image(censored, copy_id)
    logger.info("image %s %s (%s)" % (censored.region, image.name, copy_id))
    return copy_id

def finalise_image(clients, image, image_id):
    """Finalise image attributes and permissions"""
    logger.info("finalise %s %s (%s)" % (clients.region, image.name, image_id))
    req = ecs.models.ModifyImageAttributeRequest(
        region_id=clients.region,
        image_id=image_id,
        image_family=image.family,
    )
    rsp = clients.ecs.modify_image_attribute_with_options(req, RUNTIME_OPTS)
    if image.public:
        req = ecs.models.ModifyImageSharePermissionRequest(
            region_id=clients.region,
            image_id=image_id,
            is_public=True,
        )
        rsp = clients.ecs.modify_image_share_permission_with_options(
            req, RUNTIME_OPTS
        )

# Parse command-line arguments
parser = argparse.ArgumentParser(description="Import Alibaba Cloud image")
parser.add_argument('--verbose', '-v', action='count', default=0)
parser.add_argument('--name', '-n',
                    help="Base image name")
parser.add_argument('--family', '-f', default='ipxe',
                    help="Base family name")
parser.add_argument('--public', '-p', action='store_true',
                    help="Make image(s) public")
parser.add_argument('--overwrite', action='store_true',
                    help="Overwrite any existing image with same name")
parser.add_argument('--region', '-r', action='append',
                    help="AliCloud region(s)")
parser.add_argument('--role', '-R', default="AliyunFcDefaultRole",
                    help="AliCloud role for censorship bypass function")
parser.add_argument('image', nargs='+', help="iPXE disk image")
args = parser.parse_args()

# Configure logging
loglevels = [logging.WARNING, logging.INFO, logging.DEBUG]
verbosity = min(args.verbose, (len(loglevels) - 1))
logging.basicConfig(level=loglevels[verbosity])
logging.getLogger('apscheduler').setLevel(logging.WARNING)

# Use default name if none specified
if not args.name:
    args.name = '%s-%s' % (args.family,
                           datetime.date.today().strftime('%Y%m%d'))

# Use all regions if none specified
regions = args.region or all_regions()

# Construct image list
images = [image(x, args.family, args.name, args.public) for x in args.image]
imports = [(region, image) for region in regions for image in images]
workers = len(imports)

# Look up resource names
fcrole = role_arn(args.role)

# Construct per-region clients
account = account_id()
clients = {region: all_clients(region, account) for region in regions}

# Delete existing images from all regions, if applicable
if args.overwrite:
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(delete_image,
                                   clients=clients[region],
                                   name=image.name): (region, image)
                   for region, image in imports}
        done = {futures[x]: x.result() for x in as_completed(futures)}

# Delete any stale snapshots from all regions
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(delete_snapshots,
                               clients=clients[region]): region
               for region in regions}
    done = {futures[x]: x.result() for x in as_completed(futures)}

# Create temporary function in each censored region with usable FC
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(create_temp_function,
                               clients=clients[region],
                               role=fcrole): region
               for region in regions}
    funcs = {futures[x]: x.result() for x in as_completed(futures)}

# Create temporary bucket in each region with usable OSS
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(create_temp_bucket,
                               clients=clients[region],
                               func=funcs[region]): region
               for region in regions}
    buckets = {futures[x]: x.result() for x in as_completed(futures)}

# Upload image objects directly to each uncensored region with usable OSS
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(upload_object,
                               clients=clients[region],
                               bucket=buckets[region],
                               image=image): (region, image)
               for region, image in imports
               if buckets[region] and not funcs[region]}
    uploads = {futures[x]: x.result() for x in as_completed(futures)}
if not uploads:
    parser.error("At least one working non-Chinese region is required")

# Copy image objects to each censored region with usable OSS and usable FC
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(copy_objects,
                               clients=clients[region],
                               bucket=buckets[region],
                               func=funcs[region],
                               uploads=uploads): region
               for region in regions
               if buckets[region] and funcs[region]}
    done = {futures[x]: x.result() for x in as_completed(futures)}

# Import images in each region with usable OSS
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(import_image,
                               clients=clients[region],
                               image=image,
                               bucket=buckets[region]): (region, image)
               for region, image in imports if buckets[region]}
    results = {futures[x]: x.result() for x in as_completed(futures)}

# Copy images to regions without usable OSS
#
# Copies are rate-limited by source region, so spread the copies
# across all available regions with imported images.
#
copies = [(region, censored, image) for region, (censored, image) in zip(
    cycle(region for region in regions if buckets[region]),
    ((region, image) for region, image in imports if not buckets[region]),
)]
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(copy_image,
                               clients=clients[region],
                               censored=clients[censored],
                               image=image,
                               image_id=results[(region, image)]):
               (censored, image)
               for region, censored, image in copies}
    copied = {futures[x]: x.result() for x in as_completed(futures)}
    results.update(copied)

# Finalise images
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(finalise_image,
                               clients=clients[region],
                               image=image,
                               image_id=results[(region, image)]):
               (region, image)
               for region, image in imports}
    done = {futures[x]: x.result() for x in as_completed(futures)}

# Remove temporary buckets
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(delete_temp_bucket,
                               clients=clients[region],
                               func=funcs[region],
                               bucket=buckets[region]): region
               for region in regions if buckets[region]}
    done = {futures[x]: x.result() for x in as_completed(futures)}

# Remove temporary functions
with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = {executor.submit(delete_temp_function,
                               clients=clients[region],
                               func=funcs[region]): region
               for region in regions if funcs[region]}
    done = {futures[x]: x.result() for x in as_completed(futures)}

# Show created images
for region, image in imports:
    image_id = results[(region, image)]
    print("%s %s (%s) %s" % (region, image.name, image.family, image_id))