"timeout": 0,
"isPermanent": True,
"deviceId": "of:0000000000000001",
- "treatment": {
- "instructions": [
- {
- "type": "NOACTION"
- }
- ],
- "deferred": []
- },
+ "treatment": {"instructions": [{"type": "NOACTION"}], "deferred": []},
"selector": {
"criteria": [
- {
- "type": "ETH_TYPE",
- "ethType": 2048
- },
- {
- "type": "IPV4_DST",
- "ip": "10.0.0.0/32"
- }
+ {"type": "ETH_TYPE", "ethType": 2048},
+ {"type": "IPV4_DST", "ip": "10.0.0.0/32"},
]
- }
+ },
}
-flow_delete_template = {
- "deviceId": "of:0000000000000001",
- "flowId": 21392098393151996
-}
+flow_delete_template = {"deviceId": "of:0000000000000001", "flowId": 21392098393151996}
class Timer(object):
:returns req: http request object
"""
flow_list = []
- for dev_id, ip in (flows):
+ for dev_id, ip in flows:
flow = copy.deepcopy(template)
flow["deviceId"] = dev_id
- flow["selector"]["criteria"][1]["ip"] = '%s/32' % str(netaddr.IPAddress(ip))
+ flow["selector"]["criteria"][1]["ip"] = "%s/32" % str(netaddr.IPAddress(ip))
flow_list.append(flow)
body = {"flows": flow_list}
- url = 'http://' + cntl + ':' + '8181/onos/v1/flows/'
+ url = "http://" + cntl + ":" + "8181/onos/v1/flows/"
req_data = json.dumps(body)
- req = requests.Request(method, url, headers={'Content-Type': 'application/json'},
- data=req_data, auth=('onos', 'rocks'))
+ req = requests.Request(
+ method,
+ url,
+ headers={"Content-Type": "application/json"},
+ data=req_data,
+ auth=("onos", "rocks"),
+ )
return req
:returns req: http request object
"""
flow_list = []
- for dev_id, flow_id in (flows):
+ for dev_id, flow_id in flows:
flow = copy.deepcopy(template)
flow["deviceId"] = dev_id
flow["flowId"] = flow_id
flow_list.append(flow)
body = {"flows": flow_list}
- url = 'http://' + cntl + ':' + '8181/onos/v1/flows/'
+ url = "http://" + cntl + ":" + "8181/onos/v1/flows/"
req_data = json.dumps(body)
- req = requests.Request(method, url, headers={'Content-Type': 'application/json'},
- data=req_data, auth=('onos', 'rocks'))
+ req = requests.Request(
+ method,
+ url,
+ headers={"Content-Type": "application/json"},
+ data=req_data,
+ auth=("onos", "rocks"),
+ )
return req
-def _wt_request_sender(thread_id, preparefnc, inqueue=None, exitevent=None, controllers=[], restport='',
- template=None, outqueue=None, method=None):
+def _wt_request_sender(
+ thread_id,
+ preparefnc,
+ inqueue=None,
+ exitevent=None,
+ controllers=[],
+ restport="",
+ template=None,
+ outqueue=None,
+ method=None,
+):
"""The funcion sends http requests.
Runs in the working thread. It reads out flow details from the queue and sends apropriate http requests
outqueue.put(res)
-def get_device_ids(controller='127.0.0.1', port=8181):
+def get_device_ids(controller="127.0.0.1", port=8181):
"""Returns a list of switch ids"""
- rsp = requests.get(url='http://{0}:{1}/onos/v1/devices'.format(controller, port), auth=('onos', 'rocks'))
+ rsp = requests.get(
+ url="http://{0}:{1}/onos/v1/devices".format(controller, port),
+ auth=("onos", "rocks"),
+ )
if rsp.status_code != 200:
return []
- devices = json.loads(rsp.content)['devices']
- ids = [d['id'] for d in devices if 'of:' in d['id']]
+ devices = json.loads(rsp.content)["devices"]
+ ids = [d["id"] for d in devices if "of:" in d["id"]]
return ids
-def get_flow_ids(controller='127.0.0.1', port=8181):
+def get_flow_ids(controller="127.0.0.1", port=8181):
"""Returns a list of flow ids"""
- rsp = requests.get(url='http://{0}:{1}/onos/v1/flows'.format(controller, port), auth=('onos', 'rocks'))
+ rsp = requests.get(
+ url="http://{0}:{1}/onos/v1/flows".format(controller, port),
+ auth=("onos", "rocks"),
+ )
if rsp.status_code != 200:
return []
- flows = json.loads(rsp.content)['flows']
- ids = [f['id'] for f in flows]
+ flows = json.loads(rsp.content)["flows"]
+ ids = [f["id"] for f in flows]
return ids
-def get_flow_simple_stats(controller='127.0.0.1', port=8181):
+def get_flow_simple_stats(controller="127.0.0.1", port=8181):
"""Returns a list of flow ids"""
- rsp = requests.get(url='http://{0}:{1}/onos/v1/flows'.format(controller, port), auth=('onos', 'rocks'))
+ rsp = requests.get(
+ url="http://{0}:{1}/onos/v1/flows".format(controller, port),
+ auth=("onos", "rocks"),
+ )
if rsp.status_code != 200:
return []
- flows = json.loads(rsp.content)['flows']
+ flows = json.loads(rsp.content)["flows"]
res = {}
for f in flows:
- if f['state'] not in res:
- res[f['state']] = 1
+ if f["state"] not in res:
+ res[f["state"]] = 1
else:
- res[f['state']] += 1
+ res[f["state"]] += 1
return res
-def get_flow_device_pairs(controller='127.0.0.1', port=8181, flow_details=[]):
+def get_flow_device_pairs(controller="127.0.0.1", port=8181, flow_details=[]):
"""Pairing flows from controller with deteils we used ofr creation"""
- rsp = requests.get(url='http://{0}:{1}/onos/v1/flows'.format(controller, port), auth=('onos', 'rocks'))
+ rsp = requests.get(
+ url="http://{0}:{1}/onos/v1/flows".format(controller, port),
+ auth=("onos", "rocks"),
+ )
if rsp.status_code != 200:
return
- flows = json.loads(rsp.content)['flows']
+ flows = json.loads(rsp.content)["flows"]
for dev_id, ip in flow_details:
for f in flows:
# lets identify if it is our flow
item_idx = 1
else:
continue
- if f["selector"]["criteria"][item_idx]["ip"] == '%s/32' % str(netaddr.IPAddress(ip)):
+ if f["selector"]["criteria"][item_idx]["ip"] == "%s/32" % str(
+ netaddr.IPAddress(ip)
+ ):
yield dev_id, f["id"]
break
-def get_flow_to_remove(controller='127.0.0.1', port=8181):
+def get_flow_to_remove(controller="127.0.0.1", port=8181):
"""Pairing flows from controller with deteils we used ofr creation"""
- rsp = requests.get(url='http://{0}:{1}/onos/v1/flows'.format(controller, port), auth=('onos', 'rocks'))
+ rsp = requests.get(
+ url="http://{0}:{1}/onos/v1/flows".format(controller, port),
+ auth=("onos", "rocks"),
+ )
if rsp.status_code != 200:
return
- flows = json.loads(rsp.content)['flows']
+ flows = json.loads(rsp.content)["flows"]
for f in flows:
# lets identify if it is our flow
else:
continue
ipstr = f["selector"]["criteria"][item_idx]["ip"]
- if '10.' in ipstr and '/32' in ipstr:
+ if "10." in ipstr and "/32" in ipstr:
yield (f["deviceId"], f["id"])
def main(*argv):
- parser = argparse.ArgumentParser(description='Flow programming performance test: First adds and then deletes flows '
- 'into the config tree, as specified by optional parameters.')
-
- parser.add_argument('--host', default='127.0.0.1',
- help='Host where onos controller is running (default is 127.0.0.1)')
- parser.add_argument('--port', default='8181',
- help='Port on which onos\'s RESTCONF is listening (default is 8181)')
- parser.add_argument('--threads', type=int, default=1,
- help='Number of request worker threads to start in each cycle; default=1. '
- 'Each thread will add/delete <FLOWS> flows.')
- parser.add_argument('--flows', type=int, default=10,
- help='Number of flows that will be added/deleted in total, default 10')
- parser.add_argument('--fpr', type=int, default=1,
- help='Number of flows per REST request, default 1')
- parser.add_argument('--timeout', type=int, default=100,
- help='The maximum time (seconds) to wait between the add and delete cycles; default=100')
- parser.add_argument('--no-delete', dest='no_delete', action='store_true', default=False,
- help='Delete all added flows one by one, benchmark delete '
- 'performance.')
- parser.add_argument('--bulk-delete', dest='bulk_delete', action='store_true', default=False,
- help='Delete all flows in bulk; default=False')
- parser.add_argument('--outfile', default='', help='Stores add and delete flow rest api rate; default=""')
+ parser = argparse.ArgumentParser(
+ description="Flow programming performance test: First adds and then deletes flows "
+ "into the config tree, as specified by optional parameters."
+ )
+
+ parser.add_argument(
+ "--host",
+ default="127.0.0.1",
+ help="Host where onos controller is running (default is 127.0.0.1)",
+ )
+ parser.add_argument(
+ "--port",
+ default="8181",
+ help="Port on which onos's RESTCONF is listening (default is 8181)",
+ )
+ parser.add_argument(
+ "--threads",
+ type=int,
+ default=1,
+ help="Number of request worker threads to start in each cycle; default=1. "
+ "Each thread will add/delete <FLOWS> flows.",
+ )
+ parser.add_argument(
+ "--flows",
+ type=int,
+ default=10,
+ help="Number of flows that will be added/deleted in total, default 10",
+ )
+ parser.add_argument(
+ "--fpr", type=int, default=1, help="Number of flows per REST request, default 1"
+ )
+ parser.add_argument(
+ "--timeout",
+ type=int,
+ default=100,
+ help="The maximum time (seconds) to wait between the add and delete cycles; default=100",
+ )
+ parser.add_argument(
+ "--no-delete",
+ dest="no_delete",
+ action="store_true",
+ default=False,
+ help="Delete all added flows one by one, benchmark delete " "performance.",
+ )
+ parser.add_argument(
+ "--bulk-delete",
+ dest="bulk_delete",
+ action="store_true",
+ default=False,
+ help="Delete all flows in bulk; default=False",
+ )
+ parser.add_argument(
+ "--outfile",
+ default="",
+ help='Stores add and delete flow rest api rate; default=""',
+ )
in_args = parser.parse_args(*argv)
print(in_args)
base_dev_ids = get_device_ids(controller=in_args.host)
base_flow_ids = get_flow_ids(controller=in_args.host)
# ip
- ip_addr = Counter(int(netaddr.IPAddress('10.0.0.1')))
+ ip_addr = Counter(int(netaddr.IPAddress("10.0.0.1")))
# prepare func
preparefnc = _prepare_post
with Timer() as tmr:
threads = []
for i in range(int(in_args.threads)):
- thr = threading.Thread(target=_wt_request_sender, args=(i, preparefnc),
- kwargs={"inqueue": sendqueue, "exitevent": exitevent,
- "controllers": [in_args.host], "restport": in_args.port,
- "template": flow_template, "outqueue": resultqueue, "method": "POST"})
+ thr = threading.Thread(
+ target=_wt_request_sender,
+ args=(i, preparefnc),
+ kwargs={
+ "inqueue": sendqueue,
+ "exitevent": exitevent,
+ "controllers": [in_args.host],
+ "restport": in_args.port,
+ "template": flow_template,
+ "outqueue": resultqueue,
+ "method": "POST",
+ },
+ )
threads.append(thr)
thr.start()
flow_stats = get_flow_simple_stats(controller=in_args.host)
print(flow_stats)
try:
- pending_adds = int(flow_stats[u'PENDING_ADD']) # noqa # FIXME: Print this somewhere.
+ pending_adds = int(
+ flow_stats[u"PENDING_ADD"]
+ ) # noqa # FIXME: Print this somewhere.
except KeyError:
break
time.sleep(1)
if i < rounds:
print("... monitoring finished in +%d seconds\n\n" % (t.secs))
else:
- print("... monitoring aborted after %d rounds, elapsed time %d\n\n" % ((rounds, t.secs)))
+ print(
+ "... monitoring aborted after %d rounds, elapsed time %d\n\n"
+ % ((rounds, t.secs))
+ )
if in_args.no_delete:
return
with Timer() as tmr:
threads = []
for i in range(int(in_args.threads)):
- thr = threading.Thread(target=_wt_request_sender, args=(i, preparefnc),
- kwargs={"inqueue": sendqueue, "exitevent": exitevent,
- "controllers": [in_args.host], "restport": in_args.port,
- "template": flow_delete_template, "outqueue": resultqueue,
- "method": "DELETE"})
+ thr = threading.Thread(
+ target=_wt_request_sender,
+ args=(i, preparefnc),
+ kwargs={
+ "inqueue": sendqueue,
+ "exitevent": exitevent,
+ "controllers": [in_args.host],
+ "restport": in_args.port,
+ "template": flow_delete_template,
+ "outqueue": resultqueue,
+ "method": "DELETE",
+ },
+ )
threads.append(thr)
thr.start()
flow_stats = get_flow_simple_stats(controller=in_args.host)
print(flow_stats)
try:
- pending_rems = int(flow_stats[u'PENDING_REMOVE']) # noqa # FIXME: Print this somewhere.
+ pending_rems = int(
+ flow_stats[u"PENDING_REMOVE"]
+ ) # noqa # FIXME: Print this somewhere.
except KeyError:
break
time.sleep(1)
if i < rounds:
print("... monitoring finished in +%d seconds\n\n" % (t.secs))
else:
- print("... monitoring aborted after %d rounds, elapsed time %d\n\n" % ((rounds, t.secs)))
+ print(
+ "... monitoring aborted after %d rounds, elapsed time %d\n\n"
+ % ((rounds, t.secs))
+ )
if in_args.outfile != "":
- addrate = add_details['flows'] / add_details['duration']
- delrate = del_details['flows'] / del_details['duration']
+ addrate = add_details["flows"] / add_details["duration"]
+ delrate = del_details["flows"] / del_details["duration"]
print("addrate", addrate)
print("delrate", delrate)
Code Review / integration / test.git / blobdiff