Changing asyncio library to use Throttler by default?

[PhilipDAth]

Every single asyncio script I write, I have to override the semaphore to use the Thottler class, otherwise I constantly get 429 errors.  My use case is that I am processing many hundreds of async changes (could easily be 500+).

I currently use:

import throttler

	async with meraki.aio.AsyncDashboardAPI(
		output_log=False,
		print_console=False
	) as dashboard:
		dashboard._session._concurrent_requests_semaphore = throttler.Throttler(rate_limit=10, period=1.0)
		await my_super_code(dashboard)

What does everyone think about changing the actual asyncio Meraki SDK to default to the Throttler rather than using the current semaphore approach?

[sungod]

Yep, it'd get my vote.

The built-in mechanism breaks down when volume increases, over the years I've tailored scripts to reduce rate various ways depending on what they were doing.

But what I'm seeing with some customers is use of apps that simply ignore rate limiting and just pound on the API service - for example, regular storms of 50,000+ calls with fewer than 500 succeeding, the rest are 429s. This has a knock-on impact on all the well-behaved apps.

IMO it'd also be good for the API service to detect apps doing this and simply block them for a period (perhaps increasing exponentially), rather than allowing them to exhaust the call budget and block out other apps.

[rhbirkelund]

Would using Action Batches not take care of some of the calls?

[PhilipDAth]

The last job I had to do was for a customer who wanted to update an SSID across 1,280 networks.  Some networks used MRs.  Some used an MX with a WiFi module.

No templates were in use.  The SSIDs had been manually set up over a period of years and were not even in the same SSID slot in every network.

They wanted to update the networks in batches based on tags.  I had to scan every network matching the tag for the SSID and update the existing settings.

There were a lot of API calls used just to locate the SSID to be updated.  I could have considered using Action Batches for the last bit, the actual update.  I find using Action Batches a real pain.

[rhbirkelund]

I won’t say that her action batches are intuitive. It took me an awhile to get it working, but once I did it was actually okay. Depending on the call you can do either 20 or 100 at a time, so splitting into bins was necessary. I imagine you should be able to do a list with these bins and do callbacks to monitor when they are executed, and updating the queue which is just a list.

[PhilipDAth]

@Oren , is this something your team could do?  Failing that, would you consider a pull request?

[Oren]

Yep. Can you kindly provide an example of a script using aio what makes more than 10 calls/sec?

cc: @John_on_API 

[PhilipDAth]

I don't need to make more than 10 APS calls per second.  Everyone in the org running scripts and integrations probably does collectively, but that is not the core issue.

The problem is that the current async library uses a semaphore and doesn't pace the API calls out.  Consequently, you commonly run into 429 errors.

It lets the system "smash" out 10 API calls immediately - and then chokes the system.  It tends to cause 429 errors.

The Thottler() class rates limits the requests *uniformly* over time.  For example, if you configure it for 10 API calls per second, it makes sure the API calls don't happen more frequently than every 100ms.  This stops the API being smashed with the immediate 10 requests.

This stops the 429 errors from happening.

I am having to apply this fix frequently.

[PhilipDAth]

This sample code results in frequent 429 errors.  If you have more than 200 networks, it is pretty much guaranteed to fail.

#!/usr/bin/env python3
#

import os,argparse,asyncio,meraki.aio,datetime

# Load global and local Meraki settings such as MERAKI_DASHBOARD_API_KEY
from dotenv import load_dotenv
load_dotenv()
load_dotenv(dotenv_path=os.path.join(os.path.expanduser("~"),".meraki.env"))

# This function retrieves the orgId
async def getOrgId(dashboard,orgName):
	orgId=None

	# Search for the org
	for org in await dashboard.organizations.getOrganizations():
		if org['name'] == orgName:
			orgId=org['id']
			break;

	if orgId == None:
		print("Invalid organization name supplied: "+orgName)			
		exit(-1)

	return(orgId)

async def test429(dashboard,orgName):
	# Find out the organisation ID
	orgId=await getOrgId(dashboard,orgName)
	
	networks=await dashboard.organizations.getOrganizationNetworks(orgId)
	clientTasks=[dashboard.networks.getNetworkAlertsSettings(net['id']) for net in networks]
	for task in asyncio.as_completed(clientTasks):
		await task

async def main():
	# Meraki parameters
	orgName=None
	
	text="""
	429 async error generator
	"""

	parser = argparse.ArgumentParser(description = text)
	parser.add_argument("-o", "--orgName", help="Meraki org name")

	args=parser.parse_args()
	
	# Grab any arguments defined in the .env file
	orgName=os.getenv("orgName")

	# Apply any overrides from the command line
	if args.orgName: orgName=args.orgName
	
	# Check that we got all the parameters we need
	if not os.getenv("MERAKI_DASHBOARD_API_KEY"):
		print("MERAKI_DASHBOARD_API_KEY must be defined in .meraki.env in your home directory or in .env in the current directory")
		exit(-1)
	if not orgName:
		print("orgName must be defined on the command line, in .meraki.env in your home directory or in .env in the current directory")
		exit(-1)

	async with meraki.aio.AsyncDashboardAPI(
		output_log=False,
		print_console=False
	) as dashboard:
		await test429(dashboard,orgName)
	

if __name__ == "__main__":
	asyncio.run(main())

Include the throttler class and change the "with" command to the below - and it works with zero errors or warnings.

import os,argparse,asyncio,meraki.aio,datetime,throttler

	async with meraki.aio.AsyncDashboardAPI(
		output_log=False,
		print_console=False
	) as dashboard:
		dashboard._session._concurrent_requests_semaphore = throttler.Throttler(rate_limit=10, period=1.0)
		await test429(dashboard,orgName)