csit/libraries/ScaleClient.py

   1 """
   2 The purpose of this library is the ability to spread configured flows
   3 over the specified tables and switches.
   4
   5 The idea how to configure and checks inventory operational data is taken from
   6 ../../../../tools/odl-mdsal-clustering-tests/clustering-performance-test/flow_config_blaster.py
   7 ../../../../tools/odl-mdsal-clustering-tests/clustering-performance-test/inventory_crawler.py
   8 """
   9 import random
  10 import threading
  11 import netaddr
  12 import queue
  13 import requests
  14 import json
  15 import copy
  16
  17
  18 class Counter(object):
  19     def __init__(self, start=0):
  20         self.lock = threading.Lock()
  21         self.value = start
  22
  23     def increment(self, value=1):
  24         self.lock.acquire()
  25         val = self.value
  26         try:
  27             self.value += value
  28         finally:
  29             self.lock.release()
  30         return val
  31
  32
  33 _spreads = ["gauss", "linear", "first"]  # possible defined spreads at the moment
  34 _default_flow_template_json = {  # templease used for config datastore
  35     "flow": [
  36         {
  37             "hard-timeout": 65000,
  38             "idle-timeout": 65000,
  39             "cookie_mask": 4294967295,
  40             "flow-name": "FLOW-NAME-TEMPLATE",
  41             "priority": 2,
  42             "strict": False,
  43             "cookie": 0,
  44             "table_id": 0,
  45             "installHw": False,
  46             "id": "FLOW-ID-TEMPLATE",
  47             "match": {
  48                 "ipv4-destination": "0.0.0.0/32",
  49                 "ethernet-match": {"ethernet-type": {"type": 2048}},
  50             },
  51             "instructions": {
  52                 "instruction": [
  53                     {
  54                         "order": 0,
  55                         "apply-actions": {"action": [{"drop-action": {}, "order": 0}]},
  56                     }
  57                 ]
  58             },
  59         }
  60     ]
  61 }
  62
  63
  64 _node_tmpl = '/opendaylight-inventory:nodes/opendaylight-inventory:node[opendaylight-inventory:id="openflow:{0}"]'
  65
  66
  67 _default_operations_item_json = {  # template used for sal operations
  68     "input": {
  69         "bulk-flow-item": [
  70             {
  71                 "node": "to_be_replaced",
  72                 "cookie": 0,
  73                 "cookie_mask": 4294967295,
  74                 "flags": "SEND_FLOW_REM",
  75                 "hard-timeout": 65000,
  76                 "idle-timeout": 65000,
  77                 "instructions": {
  78                     "instruction": [
  79                         {
  80                             "apply-actions": {
  81                                 "action": [{"drop-action": {}, "order": 0}]
  82                             },
  83                             "order": 0,
  84                         }
  85                     ]
  86                 },
  87                 "match": {
  88                     "ipv4-destination": "0.0.0.0/32",
  89                     "ethernet-match": {"ethernet-type": {"type": 2048}},
  90                 },
  91                 "priority": 2,
  92                 "table_id": 0,
  93             }
  94         ]
  95     }
  96 }
  97
  98
  99 def _get_notes(fldet=[]):
 100     """For given list of flow details it produces a dictionary with statistics
 101     { swId1 : { tabId1 : flows_count1,
 102                 tabId2 : flows_count2,
 103                ...
 104                 'total' : switch count }
 105       swId2 ...
 106     }
 107     """
 108     notes = {}
 109     for (sw, tab, flow) in fldet:
 110         if sw not in notes:
 111             notes[sw] = {"total": 0}
 112         if tab not in notes[sw]:
 113             notes[sw][tab] = 0
 114         notes[sw][tab] += 1
 115         notes[sw]["total"] += 1
 116     return notes
 117
 118
 119 def _randomize(spread, maxn):
 120     """Returns a randomized switch or table id"""
 121     if spread not in _spreads:
 122         raise Exception("Spread method {} not available".format(spread))
 123     while True:
 124         if spread == "gauss":
 125             ga = abs(random.gauss(0, 1))
 126             rv = int(ga * float(maxn) / 3)
 127             if rv < maxn:
 128                 return rv
 129         elif spread == "linear":
 130             rv = int(random.random() * float(maxn))
 131             if rv < maxn:
 132                 return rv
 133             else:
 134                 raise ValueError("rv >= maxn")
 135         elif spread == "first":
 136             return 0
 137
 138
 139 def generate_new_flow_details(
 140     flows=10, switches=1, swspread="gauss", tables=250, tabspread="gauss"
 141 ):
 142     """Generate a list of tupples (switch_id, table_id, flow_id) which are generated
 143     according to the spread rules between swithces and tables.
 144     It also returns a dictionary with statsistics."""
 145     swflows = [_randomize(swspread, switches) for f in range(int(flows))]
 146     # we have to increse the switch index because mininet start indexing switches from 1 (not 0)
 147     fltables = [
 148         (s + 1, _randomize(tabspread, tables), idx) for idx, s in enumerate(swflows)
 149     ]
 150     notes = _get_notes(fltables)
 151     return fltables, notes
 152
 153
 154 def _prepare_add(cntl, method, flows, template=None):
 155     """Creates a PUT http requests to configure a flow in configuration datastore.
 156
 157     Args:
 158         :param cntl: controller's ip address or hostname
 159
 160         :param method: determines http request method
 161
 162         :param flows: list of flow details
 163
 164         :param template: flow template to be to be filled
 165
 166     Returns:
 167         :returns req: http request object
 168     """
 169     fl1 = flows[0]
 170     sw, tab, fl, ip = fl1
 171     url = "http://" + cntl + ":" + "8181"
 172     url += "/rests/data/opendaylight-inventory:nodes/node=openflow%3A" + str(sw)
 173     url += "/flow-node-inventory:table=" + str(tab) + "/flow=" + str(fl)
 174     flow = copy.deepcopy(template["flow"][0])
 175     flow["cookie"] = fl
 176     flow["flow-name"] = "TestFlow-%d" % fl
 177     flow["id"] = str(fl)
 178     flow["match"]["ipv4-destination"] = "%s/32" % str(netaddr.IPAddress(ip))
 179     flow["table_id"] = tab
 180     fmod = dict(template)
 181     fmod["flow"] = flow
 182     req_data = json.dumps(fmod)
 183     req = requests.Request(
 184         "PUT",
 185         url,
 186         headers={"Content-Type": "application/yang-data+json"},
 187         data=req_data,
 188         auth=("admin", "admin"),
 189     )
 190     return req
 191
 192
 193 def _prepare_table_add(cntl, method, flows, template=None):
 194     """Creates a POST http requests to configure a flow in configuration datastore.
 195
 196     Args:
 197         :param cntl: controller's ip address or hostname
 198
 199         :param method: determines http request method
 200
 201         :param flows: list of flow details
 202
 203         :param template: flow template to be to be filled
 204
 205     Returns:
 206         :returns req: http request object
 207     """
 208     fl1 = flows[0]
 209     sw, tab, fl, ip = fl1
 210     url = "http://" + cntl + ":" + "8181"
 211     url += (
 212         "/rests/data/opendaylight-inventory:nodes/node=openflow%3A"
 213         + str(sw)
 214         + "/flow-node-inventory:table="
 215         + str(tab)
 216     )
 217     fdets = []
 218     for sw, tab, fl, ip in flows:
 219         flow = copy.deepcopy(template["flow"][0])
 220         flow["cookie"] = fl
 221         flow["flow-name"] = "TestFlow-%d" % fl
 222         flow["id"] = str(fl)
 223         flow["match"]["ipv4-destination"] = "%s/32" % str(netaddr.IPAddress(ip))
 224         flow["table_id"] = tab
 225         fdets.append(flow)
 226     fmod = copy.deepcopy(template)
 227     fmod["flow"] = fdets
 228     req_data = json.dumps(fmod)
 229     req = requests.Request(
 230         "POST",
 231         url,
 232         headers={"Content-Type": "application/yang-data+json"},
 233         data=req_data,
 234         auth=("admin", "admin"),
 235     )
 236     return req
 237
 238
 239 def _prepare_delete(cntl, method, flows, template=None):
 240     """Creates a DELETE http request to remove the flow from configuration datastore.
 241
 242     Args:
 243         :param cntl: controller's ip address or hostname
 244
 245         :param method: determines http request method
 246
 247         :param flows: list of flow details
 248
 249         :param template: flow template to be to be filled
 250
 251     Returns:
 252         :returns req: http request object
 253     """
 254     fl1 = flows[0]
 255     sw, tab, fl, ip = fl1
 256     url = "http://" + cntl + ":" + "8181"
 257     url += "/rests/data/opendaylight-inventory:nodes/node=openflow%3A" + str(sw)
 258     url += "/flow-node-inventory:table=" + str(tab) + "/flow=" + str(fl)
 259     req = requests.Request(
 260         "DELETE",
 261         url,
 262         headers={"Content-Type": "application/yang-data+json"},
 263         auth=("admin", "admin"),
 264     )
 265     return req
 266
 267
 268 def _prepare_rpc_item(cntl, method, flows, template=None):
 269     """Creates a POST http requests to add or remove a flow using openflowplugin rpc.
 270
 271     Args:
 272         :param cntl: controller's ip address or hostname
 273
 274         :param method: determines http request method
 275
 276         :param flows: list of flow details
 277
 278         :param template: flow template to be to be filled
 279
 280     Returns:
 281         :returns req: http request object
 282     """
 283     f1 = flows[0]
 284     sw, tab, fl, ip = f1
 285     url = "http://" + cntl + ":" + "8181/rests/operations/sal-bulk-flow:" + method
 286     fdets = []
 287     for sw, tab, fl, ip in flows:
 288         flow = copy.deepcopy(template["input"]["bulk-flow-item"][0])
 289         flow["node"] = _node_tmpl.format(sw)
 290         flow["cookie"] = fl
 291         flow["flow-name"] = "TestFlow-%d" % fl
 292         flow["match"]["ipv4-destination"] = "%s/32" % str(netaddr.IPAddress(ip))
 293         flow["table_id"] = tab
 294         fdets.append(flow)
 295     fmod = copy.deepcopy(template)
 296     fmod["input"]["bulk-flow-item"] = fdets
 297     req_data = json.dumps(fmod)
 298     req = requests.Request(
 299         "POST",
 300         url,
 301         headers={"Content-Type": "application/yang-data+json"},
 302         data=req_data,
 303         auth=("admin", "admin"),
 304     )
 305     return req
 306
 307
 308 def _prepare_ds_item(cntl, method, flows, template=None):
 309     """Creates a POST http requests to configure a flow in configuration datastore.
 310
 311     Ofp uses write operation, standrd POST to config datastore uses read-write operation (on java level)
 312
 313     Args:
 314         :param cntl: controller's ip address or hostname
 315
 316         :param method: determines http request method
 317
 318         :param flows: list of flow details
 319
 320         :param template: flow template to be to be filled
 321
 322     Returns:
 323         :returns req: http request object
 324     """
 325     f1 = flows[0]
 326     sw, tab, fl, ip = f1
 327     url = "http://" + cntl + ":" + "8181/rests/operations/sal-bulk-flow:" + method
 328     fdets = []
 329     for sw, tab, fl, ip in flows:
 330         flow = copy.deepcopy(template["input"]["bulk-flow-item"][0])
 331         flow["node"] = _node_tmpl.format(sw)
 332         flow["cookie"] = fl
 333         flow["flow-name"] = "TestFlow-%d" % fl
 334         flow["match"]["ipv4-destination"] = "%s/32" % str(netaddr.IPAddress(ip))
 335         flow["table_id"] = tab
 336         flow["flow-id"] = fl
 337         fdets.append(flow)
 338     fmod = copy.deepcopy(template)
 339     del fmod["input"]["bulk-flow-item"]
 340     fmod["input"]["bulk-flow-ds-item"] = fdets
 341     req_data = json.dumps(fmod)
 342     req = requests.Request(
 343         "POST",
 344         url,
 345         headers={"Content-Type": "application/yang-data+json"},
 346         data=req_data,
 347         auth=("admin", "admin"),
 348     )
 349     return req
 350
 351
 352 def _wt_request_sender(
 353     thread_id,
 354     preparefnc,
 355     inqueue=None,
 356     exitevent=None,
 357     controllers=[],
 358     restport="",
 359     template=None,
 360     outqueue=None,
 361     method=None,
 362 ):
 363     """The funcion sends http requests.
 364
 365     Runs in the working thread. It reads out flow details from the queue and sends apropriate http requests
 366     to the controller
 367
 368     Args:
 369         :param thread_id: thread id
 370
 371         :param preparefnc: function to preparesthe http request
 372
 373         :param inqueue: input queue, flow details are comming from here
 374
 375         :param exitevent: event to notify working thread that parent (task executor) stopped filling the input queue
 376
 377         :param controllers: a list of controllers' ip addresses or hostnames
 378
 379         :param restport: restconf port
 380
 381         :param template: flow template used for creating flow content
 382
 383         :param outqueue: queue where the results should be put
 384
 385         :param method: method derermines the type of http request
 386
 387     Returns:
 388         nothing, results must be put into the output queue
 389     """
 390     ses = requests.Session()
 391     cntl = controllers[0]
 392     counter = [0 for i in range(600)]
 393     loop = True
 394     req_no = 0
 395     num_errors = 0
 396
 397     while loop:
 398         req_no += 1
 399         try:
 400             flowlist = inqueue.get(timeout=1)
 401         except queue.Empty:
 402             if exitevent.is_set() and inqueue.empty():
 403                 loop = False
 404             continue
 405         req = preparefnc(cntl, method, flowlist, template=template)
 406         # prep = ses.prepare_request(req)
 407         prep = req.prepare()
 408         try:
 409             rsp = ses.send(prep, timeout=5)
 410         except requests.exceptions.Timeout:
 411             print(f"*WARN* [{req_no}] Timeout: {req.method} {req.url}")
 412             counter[99] += 1
 413             if counter[99] > 10:
 414                 print("*ERROR* Too many timeouts.")
 415                 break
 416             continue
 417         else:
 418             if rsp.status_code not in [200, 201, 204]:
 419                 print(
 420                     f"*WARN* [{req_no}] Status code {rsp.status_code}:"
 421                     f" {req.method} {req.url}\n{rsp.text}"
 422                 )
 423                 num_errors += 1
 424                 if num_errors > 10:
 425                     print("*ERROR* Too many errors.")
 426                     break
 427         counter[rsp.status_code] += 1
 428     res = {}
 429     for i, v in enumerate(counter):
 430         if v > 0:
 431             res[i] = v
 432     outqueue.put(res)
 433
 434
 435 def _task_executor(
 436     method="",
 437     flow_template=None,
 438     flow_details=[],
 439     controllers=["127.0.0.1"],
 440     restport="8181",
 441     nrthreads=1,
 442     fpr=1,
 443 ):
 444     """The main function which drives sending of http requests.
 445
 446     Creates 2 queues and requested number of 'working threads'.  One queue is filled with flow details and working
 447     threads read them out and send http requests. The other queue is for sending results from working threads back.
 448     After the threads' join, it produces a summary result.
 449
 450     Args:
 451         :param method: based on this the function which prepares http request is choosen
 452
 453         :param flow_template: template to generate a flow content
 454
 455         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 456
 457         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 458
 459         :param restport: restconf port (default='8181')
 460
 461         :param nrthreads: number of threads used to send http requests (default=1)
 462
 463         :param fpr: flow per request, number of flows sent in one http request
 464
 465     Returns:
 466         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 467     """
 468     # TODO: multi controllers support
 469     ip_addr = Counter(int(netaddr.IPAddress("10.0.0.1")))
 470
 471     # choose message prepare function
 472     if method == "PUT":
 473         preparefnc = _prepare_add
 474         # put can contain only 1 flow, lets overwrite any value of flows per request
 475         fpr = 1
 476     elif method == "POST":
 477         preparefnc = _prepare_table_add
 478     elif method == "DELETE":
 479         preparefnc = _prepare_delete
 480         # delete flow can contain only 1 flow, lets overwrite any value of flows per request
 481         fpr = 1
 482     elif method in ["add-flows-ds", "remove-flows-ds"]:
 483         preparefnc = _prepare_ds_item
 484     elif method in ["add-flows-rpc", "remove-flows-rpc"]:
 485         preparefnc = _prepare_rpc_item
 486     else:
 487         raise NotImplementedError(
 488             "Method {0} does not have it's prepeare function defined".format(method)
 489         )
 490
 491     # lets enlarge the tupple of flow details with IP, to be used with the template
 492     flows = [(sw, tab, flo, ip_addr.increment()) for sw, tab, flo in flow_details]
 493     # lels divide flows into switches and tables - flow groups
 494     flowgroups = {}
 495     for flow in flows:
 496         sw, tab, _, _ = flow
 497         flowkey = (sw, tab)
 498         if flowkey in flowgroups:
 499             flowgroups[flowkey].append(flow)
 500         else:
 501             flowgroups[flowkey] = [flow]
 502
 503     # lets fill the queue with details needed for one http requests
 504     # we have lists with flow details for particular (switch, table) tupples, now we need to split the lists
 505     # according to the flows per request (fpr) paramer
 506     sendqueue = queue.Queue()
 507     for flowgroup, flow_list in flowgroups.items():
 508         while len(flow_list) > 0:
 509             sendqueue.put(flow_list[: int(fpr)])
 510             flow_list = flow_list[int(fpr) :]
 511
 512     # result_gueue
 513     resultqueue = queue.Queue()
 514     # creaet exit event
 515     exitevent = threading.Event()
 516
 517     # lets start threads whic will read flow details fro queues and send
 518     threads = []
 519     for i in range(int(nrthreads)):
 520         thr = threading.Thread(
 521             target=_wt_request_sender,
 522             args=(i, preparefnc),
 523             kwargs={
 524                 "inqueue": sendqueue,
 525                 "exitevent": exitevent,
 526                 "controllers": controllers,
 527                 "restport": restport,
 528                 "template": flow_template,
 529                 "outqueue": resultqueue,
 530                 "method": method,
 531             },
 532         )
 533         threads.append(thr)
 534         thr.start()
 535
 536     exitevent.set()
 537
 538     result = {}
 539     # waitng for reqults and sum them up
 540     for t in threads:
 541         t.join()
 542         # reading partial resutls from sender thread
 543         part_result = resultqueue.get()
 544         for k, v in part_result.items():
 545             if k not in result:
 546                 result[k] = v
 547             else:
 548                 result[k] += v
 549     return result
 550
 551
 552 def configure_flows(*args, **kwargs):
 553     """Configure flows based on given flow details.
 554
 555     Args:
 556         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 557
 558         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 559
 560         :param restport: restconf port (default='8181')
 561
 562         :param nrthreads: number of threads used to send http requests (default=1)
 563
 564     Returns:
 565         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 566     """
 567     return _task_executor(
 568         method="PUT", flow_template=_default_flow_template_json, **kwargs
 569     )
 570
 571
 572 def deconfigure_flows(*args, **kwargs):
 573     """Deconfigure flows based on given flow details.
 574
 575     Args:
 576         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 577
 578         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 579
 580         :param restport: restconf port (default='8181')
 581
 582         :param nrthreads: number of threads used to send http requests (default=1)
 583
 584     Returns:
 585         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 586     """
 587     return _task_executor(
 588         method="DELETE", flow_template=_default_flow_template_json, **kwargs
 589     )
 590
 591
 592 def configure_flows_bulk(*args, **kwargs):
 593     """Configure flows based on given flow details using a POST http request..
 594
 595     Args:
 596         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 597
 598         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 599
 600         :param restport: restconf port (default='8181')
 601
 602         :param nrthreads: number of threads used to send http requests (default=1)
 603
 604     Returns:
 605         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 606     """
 607     return _task_executor(
 608         method="POST", flow_template=_default_flow_template_json, **kwargs
 609     )
 610
 611
 612 def operations_add_flows_ds(*args, **kwargs):
 613     """Configure flows based on given flow details.
 614
 615     Args:
 616         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 617
 618         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 619
 620         :param restport: restconf port (default='8181')
 621
 622         :param nrthreads: number of threads used to send http requests (default=1)
 623
 624     Returns:
 625         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 626     """
 627     return _task_executor(
 628         method="add-flows-ds", flow_template=_default_operations_item_json, **kwargs
 629     )
 630
 631
 632 def operations_remove_flows_ds(*args, **kwargs):
 633     """Remove flows based on given flow details.
 634
 635     Args:
 636         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 637
 638         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 639
 640         :param restport: restconf port (default='8181')
 641
 642         :param nrthreads: number of threads used to send http requests (default=1)
 643
 644     Returns:
 645         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 646     """
 647     return _task_executor(
 648         method="remove-flows-ds", flow_template=_default_operations_item_json, **kwargs
 649     )
 650
 651
 652 def operations_add_flows_rpc(*args, **kwargs):
 653     """Configure flows based on given flow details using rpc calls.
 654
 655     Args:
 656         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 657
 658         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 659
 660         :param restport: restconf port (default='8181')
 661
 662         :param nrthreads: number of threads used to send http requests (default=1)
 663
 664     Returns:
 665         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 666     """
 667     return _task_executor(
 668         method="add-flows-rpc", flow_template=_default_operations_item_json, **kwargs
 669     )
 670
 671
 672 def operations_remove_flows_rpc(*args, **kwargs):
 673     """Remove flows based on given flow details using rpc calls.
 674
 675     Args:
 676         :param flow_details: a list of tupples with flow details (switch_id, table_id, flow_id, ip_addr) (default=[])
 677
 678         :param controllers: a list of controllers host names or ip addresses (default=['127.0.0.1'])
 679
 680         :param restport: restconf port (default='8181')
 681
 682         :param nrthreads: number of threads used to send http requests (default=1)
 683
 684     Returns:
 685         :returns dict: dictionary of http response counts like {'http_status_code1: 'count1', etc.}
 686     """
 687     return _task_executor(
 688         method="remove-flows-rpc", flow_template=_default_operations_item_json, **kwargs
 689     )
 690
 691
 692 def _get_operational_inventory_of_switches(controller):
 693     """Gets number of switches present in the operational inventory
 694
 695     Args:
 696         :param controller: controller's ip or host name
 697
 698     Returns:
 699         :returns switches: number of switches connected
 700     """
 701     url = (
 702         "http://"
 703         + controller
 704         + ":8181/rests/data/opendaylight-inventory:nodes?content=nonconfig"
 705     )
 706     rsp = requests.get(
 707         url,
 708         headers={"Accept": "application/yang-data+json"},
 709         stream=False,
 710         auth=("admin", "admin"),
 711     )
 712     if rsp.status_code != 200:
 713         return None
 714     inv = json.loads(rsp.content)
 715     if "opendaylight-inventory:nodes" not in inv:
 716         return None
 717     if "node" not in inv["opendaylight-inventory:nodes"]:
 718         return []
 719     inv = inv["opendaylight-inventory:nodes"]["node"]
 720     switches = [sw for sw in inv if "openflow:" in sw["id"]]
 721     return switches
 722
 723
 724 def flow_stats_collected(controller=""):
 725     """Provides the operational inventory counts counts of switches and flows.
 726
 727     Args:
 728         :param controller: controller's ip address or host name
 729
 730     Returns:
 731         :returns (switches, flows_reported, flows-found): tupple with counts of switches, reported and found flows
 732     """
 733     active_flows = 0
 734     found_flows = 0
 735     switches = _get_operational_inventory_of_switches(controller)
 736     if switches is None:
 737         return 0, 0, 0
 738     for sw in switches:
 739         tabs = sw["flow-node-inventory:table"]
 740         for t in tabs:
 741             active_flows += t[
 742                 "opendaylight-flow-table-statistics:flow-table-statistics"
 743             ]["active-flows"]
 744             if "flow" in t:
 745                 found_flows += len(t["flow"])
 746     print(
 747         (
 748             "Switches,ActiveFlows(reported)/FlowsFound",
 749             len(switches),
 750             active_flows,
 751             found_flows,
 752         )
 753     )
 754     return len(switches), active_flows, found_flows
 755
 756
 757 def get_switches_count(controller=""):
 758     """Gives the count of the switches presnt in the operational inventory nodes datastore.
 759
 760     Args:
 761         :param controller: controller's ip address or host name
 762
 763     Returns:
 764         :returns switches: returns the number of connected switches
 765     """
 766     switches = _get_operational_inventory_of_switches(controller)
 767     if switches is None:
 768         return 0
 769     return len(switches)
 770
 771
 772 def validate_responses(received, expected):
 773     """Compares given response summary with expected results.
 774
 775     Args:
 776         :param received: dictionary returned from operations_* and (de)configure_flows*
 777                          of this library
 778                          e.g. received = { 200:41 } - this means that we 41x receives response with status code 200
 779
 780         :param expected: list of expected http result codes
 781                          e.g. expected=[200] - we expect only http status 200 to be present
 782
 783     Returns:
 784         :returns True: if list of http statuses from received responses is the same as exxpected
 785         :returns False: elseware
 786     """
 787     return True if list(received.keys()) == expected else False