How to use Retry in AsyncIO properly?

[otodorov]

Hello,
I'm trying to create a google cloud function that will scale my clusters based on specific tag.

I made a function def list_gke_clusters(label: str) -> dict: that acpet a label as a string and return dict like:

{
    '<cluster_name>-<node_pool_name>': {
        "cluster_name": "<cluster_name>",
        "location": "<location>",
        "labels": "{'key1': 'val1', 'key2': 'val2'}",
        "node_pool": "<node_pool_name>"
    }
}

Then I made a second function that is used to scale the cluster async def resize_gke_node_pool(cluster_dict: dict, node_number: int): by following the docs Retry in AsyncIO
The function accepts clusters as a dict and node_number as int to set desired cluster size.

Here is the whole function:

async def resize_gke_node_pool(cluster_dict: dict, node_number: int):****
    '''
    Args:
        cluster_name (string): The name of the cluster that owns the node pool
        location (string): The name of the Google Compute Engine zone|region
        pool (string): The name of the node pool to set size
        node_number (intiger): The desired node count for the pool.

    Return:
        cluster_name (string): The name of the cluster that owns the node pool
        node pool (string): The name of the node pool size
        location (string): The name of the Google Compute Engine zone|region
        time (string): The time when the execution started

    name string format: "projects/<project>/locations/<location>/clusters/<cluster_name>/nodePools/<node_pool>"
    '''

    request = container_v1.SetNodePoolSizeRequest(
        name=f"projects/{project}/locations/{cluster_dict['location']}/clusters/{cluster_dict['cluster_name']}/nodePools/{cluster_dict['node_pool']}",
        node_count=node_number,
    )

    response = await client.set_node_pool_size(
        request=request,
        retry=retry_async.AsyncRetry(
            initial=1.0,
            deadline=900.0,
            predicate=retry_async.if_exception_type(
                exceptions.FailedPrecondition)
        )
    )

    print(f"""
    cluster_name: {cluster_dict['cluster_name']}
    node pool: {cluster_dict['node_pool']} is set to {node_number}
    location: {cluster_dict['location']}
    """)

    return response

This is an example of cluster_dict passed to resize_gke_node_pool:

{'cluster_name': 'cluster1', 'location': 'europe-west1', 'labels': {'env': 'test', 'scale': 'true'}, 'node_pool': 'cluster1-pool'}
{'cluster_name': 'cluster1', 'location': 'europe-west1', 'labels': {'env': 'test', 'scale': 'true'}, 'node_pool': 'pool-1'}
{'cluster_name': 'cluster2', 'location': 'europe-west1-b', 'labels': {'env': 'test', 'scale': 'true'}, 'node_pool': 'cluster2-pool'}
{'cluster_name': 'cluster2', 'location': 'europe-west1-b', 'labels': {'env': 'test', 'scale': 'true'}, 'node_pool': 'pool-1'}

execute function used to send data to resize_gke_node_pool

async def execute(event):
    if 'attributes' in event:
        # CLusters that have this labels will be targeted
        label = event['attributes']['label']
        # Set desrired number for each Cluster Node Pool
        node_number = int(event['attributes']['node_number'])

        cluster_list = [list_gke_clusters(label)[cluster]
                        for cluster in list_gke_clusters(label)]

        await asyncio.gather(*[resize_gke_node_pool(cluster, node_number) for cluster in cluster_list])

and the main function:

def main(event, context):
    asyncio.run(execute(event))

What I want to achieve is to scale all clusters that match specific label simultaniously.
Some of the cluster has more then one node pool, so when a spcific node pool in a particular cluster is updated, the rest of the node pools on that cluster should retry until the cluster is editable.

When I execute the script, I receive the following error:

Traceback (most recent call last):
  File "/git/GCP_CloudFunctions/gke_cluster_resize/gke_cluster_resize.py", line 118, in <module>
    main(event, context=None)
  File "/git/GCP_CloudFunctions/gke_cluster_resize/gke_cluster_resize.py", line 108, in main
    asyncio.run(execute(event))
  File "/root/.pyenv/versions/3.9.2/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/root/.pyenv/versions/3.9.2/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/git/GCP_CloudFunctions/gke_cluster_resize/gke_cluster_resize.py", line 104, in execute
    await asyncio.gather(*[resize_gke_node_pool(cluster, node_number) for cluster in cluster_list])
  File "/git/GCP_CloudFunctions/gke_cluster_resize/gke_cluster_resize.py", line 75, in resize_gke_node_pool
    response = await client.set_node_pool_size(
  File "/root/.venv3.9.2-gke-cluster-resize/lib/python3.9/site-packages/google/api_core/retry_async.py", line 215, in retry_wrapped_func
    return await retry_target(
  File "/root/.venv3.9.2-gke-cluster-resize/lib/python3.9/site-packages/google/api_core/retry_async.py", line 115, in retry_target
    return await asyncio.wait_for(
  File "/root/.pyenv/versions/3.9.2/lib/python3.9/asyncio/tasks.py", line 462, in wait_for
    fut = ensure_future(fut, loop=loop)
  File "/root/.pyenv/versions/3.9.2/lib/python3.9/asyncio/tasks.py", line 679, in ensure_future
    raise TypeError('An asyncio.Future, a coroutine or an awaitable is '
TypeError: An asyncio.Future, a coroutine or an awaitable is required

P.S. I achieved what I wanted by using the multiprocessing ThreadPool, but I believe using asyncio will be better.

[otodorov]

Once I changed the function to use a decorator, It does work.