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.