EDL Latency Poking

edl_latency_test.py

import os
import time
import json
import argparse
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from statistics import mean
from random import randrange

MIN_DELAY=0.5
MAX_DELAY=10
ATTEMPTS=2

parser = argparse.ArgumentParser()
parser.add_argument('-a','--appcreds', dest="app_creds", default=os.getenv('APP_CREDS'), help="Base64 Encoded EDL App UID/PWD")
parser.add_argument('-c','--clientid', dest="client_id", default=os.getenv('CLIENT_ID'), help="EDL App Client ID")
parser.add_argument('-t','--edltoken', dest="edl_token", default=os.getenv('EDL_TOKEN'), help="EDL Token")
parser.add_argument('--mindelay', dest="min_delay", default=MIN_DELAY, help="Minimum time to wait between requests (fractional seconds)", type=float)
parser.add_argument('--maxdelay', dest="max_delay", default=MAX_DELAY, help="Maximum time to wait between requests (fractional seconds)", type=float)
parser.add_argument('--attempts', dest="attempts", default=ATTEMPTS, help="Number of requests to make of each endpoint", type=int)
args = parser.parse_args()

# POST data
headers = { "Authorization": f"Basic {args.app_creds}" }
data = urlencode( {"client_id": args.client_id, "token": args.edl_token } ).encode('utf8')

edl_hosts = {
  'onprem': {'url': 'https://urs.earthdata.nasa.gov', 'latency':[]  } ,
  'cloud':  {'url': 'https://alternative.cloudfront.net', 'latency':[] } 
}

def make_request ( url ):
    timer = time.time()
    req = Request( url, data=data, headers=headers)
    try: 
        resp = urlopen(req).read().decode('utf8')
        user_id = json.loads(resp)['uid']
    except Exception as E:
        print(f"Could not hit {url}: {E}")
    
    # Return our time a milliseconds
    duration = time.time() - timer
    return(float("{:.2f}".format(duration*1000)))

for target,host in edl_hosts.items():
    for attempt in range (0, args.attempts):
        print(f" > Making request {(1+attempt)} of {args.attempts} for {target}")
        latency = make_request ( host['url'] + '/oauth/tokens/user' )

        # record messured latency
        edl_hosts[target]['latency'].append(latency)
        delay = randrange(args.min_delay*10,args.max_delay*10)/10 

        print(f" > Request latency was {latency}, sleeping for {delay}")
        time.sleep(delay)

    avg_latency = mean(edl_hosts[target]['latency'])
    print(f"Average request latency for {target} was {avg_latency:.2f}ms")