Make Thousands of NSX-T API Calls in Minute

Background

As a network guy, network automation has always been my favourite topic. I needed to build an NSX-T environment to validate a specific function from time to time. When I got an NSX-T config, I used a straightforward Python script that makes the NSX-T API calls to import the config into a new lab SDDC. My script works well usually and can complete the job in a reasonable time (e.g. 5 min to 10 mins) if the config is not too complicated: dozens of network segments, 100 NSX-T gateway firewall rules, or DFW rules. But when I tried to test the DFW scalability, I found that my simple script didn’t work as desired. Importing a large-scale SDDC NSX-T config (including thousands of groups, services and rules) usually takes 2 hours or more. It sometimes can’t complete the import task due to an expired token(of course, the issue could be avoided if I had a better token renewal). The challenges stemmed from the Python request module I used: the Python request module can only make synchronous requests, which means the module must wait for the previous API call to be completed before making a new API request.

I hate waiting, especially when the result is a failure. So I made up my mind to find a way that can make thousands of NSX-T API calls in a minute.

Solution

After a bit of research, I found that the following 2 Python modules can help me out.

• AioHTTP: Asynchronous HTTP Client/Server for asyncio and Python.
• AsyncIO: The library for writing single-threaded concurrent code using coroutines, multiplexing I/O access over sockets and other resources, running network clients and servers, and other related primitives.

My code includes 4 parts:

Part 1: Import required Python modules and initialize required variables.

import json
import os
import sys
import string
import time
import requests
import aiohttp
import asyncio
from aiohttp import ClientSession
from requests.exceptions import HTTPError

cwd = os.getcwd() # get current working directory

# Open the input para file. The para file includes refresh token and NSX manager proxy URL
inputFile = cwd + "/" + sys.argv[1]

try:
    with open(inputFile) as filehandle:
        paras = json.load(filehandle) 
except:
    print('Import failed - unable to open',inputFile)

proxy_url = paras["proxy_url"]
myrefresh_token = paras["myrefresh_token"]

# define the logfile
date = time.strftime("%Y-%m-%d-%H-%M-%S")
log_file = cwd + "/" + "asyncio_run-"+date +".txt"

Part 2: Get VMC SDDC Access Token

api_url = "https://console.cloud.vmware.com/csp/gateway/am/api/auth/api-tokens/authorize"
def getAccessToken(refresh_token):
        """ Gets the Access Token using the Refresh Token """
        params = {'refresh_token': refresh_token}
        headers = {'Content-Type': 'application/json'}
        response = requests.post(url=api_url, params=params, headers=headers)
        jsonResponse = response.json()
        try:
            access_token = jsonResponse['access_token']
        except:
            access_token = ""
        return access_token
access_token = getAccessToken(myrefresh_token)
# Authentication header
myHeader = {"Content-Type": "application/json","Accept": "application/json", 'csp-auth-token': access_token }

Part 3: Define functions to add NSX-T service from a service list

f = open(log_file, "a")

async def addservice(session,service, sema):
    nsxm_url = proxy_url + "/policy/api/v1/infra/services/" + service["id"]
    try:
        async with sema:
            response = await session.request(method='Patch', url=nsxm_url, headers=myHeader, json=service, ssl=False)
            #response.raise_for_status()
            f.write(f"Response status ({nsxm_url}): {response.status}\n")
    except HTTPError as http_err:
        f.write(f"HTTP error occurred: {http_err}\n")
    except Exception as err:
        f.write(f"An error ocurred: {err}\n")

async def main():
    sema = asyncio.Semaphore(40)
    tasks = []
    async with aiohttp.ClientSession() as session:    
        for service in services_list:
            tasks.append(addservice(session,service, sema))
        await asyncio.gather(*tasks)
 # asyncio.Semaphore(x) is used to limit the total number of connections (40 in my case) to NSX-T manager.

Note: VMC-AWS NSX limits the allowed concurrent session to 40. You possibly want to reduce the semaphore limit from 40 to 30; that ensures you always stay safe.

Part 4: Call the async main function

async def main():
    sema = asyncio.Semaphore(40)
    tasks = []
    async with aiohttp.ClientSession() as session:    
        for service in services_list:
            tasks.append(addservice(session,service, sema))
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(main())
    print("--- %s seconds ---" % (time.time() - start_time))

This is the end of this blog. Thank you for reading!

Update: I have just added an example of the input file and service_list file in my Github repo. Feel free to check them out.

Originally published at http://davidwzhang.com on October 3, 2023.