debug_snapshot.py 文件源码

def get_container_id_mapping(pool, compose_cmd):
    service_names = subprocess.check_output(
        compose_cmd + ["config", "--services"]
    )
    service_names = service_names.strip().decode("utf-8").split("\n")
    id_mapping = {
        name: pool.apply_async(pool_container_id, (name, compose_cmd))
        for name in service_names
    }

    while not all(future.ready() for future in id_mapping.values()):
        time.sleep(0.1)
    for name, future in list(id_mapping.items()):
        if not future.successful():
            raise RuntimeError("Cannot get ID of service {0}".format(name))
        id_mapping[name] = future.get()

    return id_mapping