tools/odl-mdsal-clustering-tests/clustering-performance-test/shard_perf_test.py

   1 #!/usr/bin/python
   2 from random import randrange
   3 import json
   4 import argparse
   5 import time
   6 import threading
   7 import sys
   8 import requests
   9
  10
  11 __author__ = "Jan Medved"
  12 __copyright__ = "Copyright(c) 2014, Cisco Systems, Inc."
  13 __license__ = "New-style BSD"
  14 __email__ = "jmedved@cisco.com"
  15
  16
  17 class Counter(object):
  18     def __init__(self, start=0):
  19         self.lock = threading.Lock()
  20         self.value = start
  21
  22     def increment(self, value=1):
  23         self.lock.acquire()
  24         val = self.value
  25         try:
  26             self.value += value
  27         finally:
  28             self.lock.release()
  29         return val
  30
  31
  32 class Timer(object):
  33     def __init__(self, verbose=False):
  34         self.verbose = verbose
  35
  36     def __enter__(self):
  37         self.start = time.time()
  38         return self
  39
  40     def __exit__(self, *args):
  41         self.end = time.time()
  42         self.secs = self.end - self.start
  43         self.msecs = self.secs * 1000  # millisecs
  44         if self.verbose:
  45             print("elapsed time: %f ms" % self.msecs)
  46
  47
  48 class ShardPerformanceTester(object):
  49     """
  50     The ShardPerformanceTester class facilitates performance testing of CDS shards. The test starts a number of
  51     threads, where each thread issues a specified number of resource retrieval requests to URLs specified at the
  52     beginning of a test. A ShardPerformanceTester object gets its connection parameters to the system-under-test
  53     and its test parameters when it is instantiated.  The set of URLs from where to retrieve resources is
  54     specified when a test is started. By passing in the appropriate URLs, the test can be used to test data
  55     retrieval performance of different shards or different resources at different granularities, etc.
  56     """
  57
  58     headers = {"Accept": "application/json"}
  59
  60     def __init__(self, host, port, auth, threads, nrequests, plevel):
  61         """"""
  62         self.host = host
  63         self.port = port
  64         self.auth = auth
  65         self.requests = nrequests
  66         self.threads = threads
  67         self.plevel = plevel
  68
  69         self.print_lock = threading.Lock()
  70         self.cond = threading.Condition()
  71         self.threads_done = 0
  72
  73         self.ok_requests = 0
  74         self.url_counters = []
  75         self.total_rate = 0
  76
  77     def make_request(self, session, urls):
  78         """
  79         Makes a request for a resource at a random URL selected from a list of URLs passed as input parameter
  80         :param session: Session to system under test
  81         :param urls: List of resource URLs
  82         :return: Status code from the resource request call
  83         """
  84         url_index = randrange(0, len(urls))
  85         r_url = urls[url_index]
  86         self.url_counters[url_index].increment()
  87
  88         if not self.auth:
  89             r = session.get(r_url, headers=self.headers, stream=False)
  90         else:
  91             r = session.get(
  92                 r_url, headers=self.headers, stream=False, auth=("admin", "admin")
  93             )
  94         return r.status_code
  95
  96     def worker(self, tid, urls):
  97         """
  98         Worker thread function. Connects to system-under-test and makes 'self.requests' requests for
  99         resources to URLs randomly selected from 'urls'
 100         :param tid: Worker thread ID
 101         :param urls: List of resource URLs
 102         :return: None
 103         """
 104         res = {200: 0}
 105
 106         s = requests.Session()
 107
 108         with self.print_lock:
 109             print("    Thread %d: Performing %d requests" % (tid, self.requests))
 110
 111         with Timer() as t:
 112             for r in range(self.requests):
 113                 sts = self.make_request(s, urls)
 114                 try:
 115                     res[sts] += 1
 116                 except KeyError:
 117                     res[sts] = 1
 118
 119         ok_rate = res[200] / t.secs
 120         total_rate = sum(res.values()) / t.secs
 121
 122         with self.print_lock:
 123             print("Thread %d done:" % tid)
 124             print("    Time: %.2f," % t.secs)
 125             print("    Success rate:  %.2f, Total rate: %.2f" % (ok_rate, total_rate))
 126             print("    Per-thread stats: ")
 127             print(res)
 128             self.threads_done += 1
 129             self.total_rate += total_rate
 130
 131         s.close()
 132
 133         with self.cond:
 134             self.cond.notifyAll()
 135
 136     def run_test(self, urls):
 137         """
 138         Runs the performance test. Starts 'self.threads' worker threads, waits for all of them to finish and
 139         prints results.
 140         :param urls: List of urls from which to request resources
 141         :return: None
 142         """
 143
 144         threads = []
 145         self.total_rate = 0
 146
 147         # Initialize url counters
 148         del self.url_counters[:]
 149         for i in range(len(urls)):
 150             self.url_counters.append(Counter(0))
 151
 152         # Start all worker threads
 153         for i in range(self.threads):
 154             t = threading.Thread(target=self.worker, args=(i, urls))
 155             threads.append(t)
 156             t.start()
 157
 158         # Wait for all threads to finish and measure the execution time
 159         with Timer() as t:
 160             while self.threads_done < self.threads:
 161                 with self.cond:
 162                     self.cond.wait()
 163
 164         # Print summary results. Each worker prints its owns results too.
 165         print("\nSummary Results:")
 166         print(
 167             "    Requests/sec (total_sum): %.2f"
 168             % ((self.threads * self.requests) / t.secs)
 169         )
 170         print(
 171             "    Requests/sec (measured):  %.2f"
 172             % ((self.threads * self.requests) / t.secs)
 173         )
 174         print("    Time: %.2f" % t.secs)
 175         self.threads_done = 0
 176
 177         if self.plevel > 0:
 178             print("    Per URL Counts: ")
 179             for i in range(len(urls)):
 180                 print("%d" % self.url_counters[i].value)
 181             print("\n")
 182
 183
 184 class TestUrlGenerator(object):
 185     """
 186     Base abstract class to generate test URLs for ShardPerformanceTester. First, an entire subtree representing
 187     a shard or a set of resources is retrieved, then a set of URLS to access small data stanzas is created. This
 188     class only defines the framework, the methods that create URL sets are defined in derived classes.
 189     """
 190
 191     def __init__(self, host, port, auth):
 192         """
 193         Initialization
 194         :param host: Controller's IP address
 195         :param port: Controller's RESTCONF port
 196         :param auth: Indicates whether to use authentication with default user/password (admin/admin)
 197         :return: None
 198         """
 199         self.host = host
 200         self.port = port
 201         self.auth = auth
 202         self.resource_string = ""
 203
 204     def url_generator(self, data):
 205         """
 206         Abstract  URL generator. Must be overridden in a derived class
 207         :param data: Bulk resource data (JSON) from which to generate the URLs
 208         :return: List of generated Resources
 209         """
 210         print(
 211             "Abstract class '%s' should never be used standalone"
 212             % (self.__class__.__name__)
 213         )
 214         return []
 215
 216     def generate(self):
 217         """
 218         Drives the generation of test URLs. First, it gets a 'bulk' resource (e.g. the entire inventory
 219          or the entire topology) from the controller specified during int()  and then invokes a resource-specific
 220          URL generator to create a set of resource-specific URLs.
 221         """
 222         t_url = "http://" + self.host + ":" + self.port + "/" + self.resource_string
 223         headers = {"Accept": "application/json"}
 224         r_url = []
 225
 226         if not self.auth:
 227             r = requests.get(t_url, headers=headers, stream=False)
 228         else:
 229             r = requests.get(
 230                 t_url, headers=headers, stream=False, auth=("admin", "admin")
 231             )
 232
 233         if r.status_code != 200:
 234             print(
 235                 "Failed to get HTTP response from '%s', code %d"
 236                 % ((t_url, r.status_code))
 237             )
 238         else:
 239             try:
 240                 r_url = self.url_generator(json.loads(r.content))
 241             except Exception:
 242                 print(
 243                     "Failed to get json from '%s'. Please make sure you are connected to mininet."
 244                     % (r_url)
 245                 )
 246
 247         return r_url
 248
 249
 250 class TopoUrlGenerator(TestUrlGenerator):
 251     """
 252     Class to generate test URLs from the topology shard.
 253     :return: List of generated Resources
 254     """
 255
 256     def __init__(self, host, port, auth):
 257         TestUrlGenerator.__init__(self, host, port, auth)
 258         self.resource_string = (
 259             "restconf/operational/network-topology:network-topology/topology/flow:1"
 260         )
 261
 262     def url_generator(self, topo_data):
 263         url_list = []
 264         try:
 265             nodes = topo_data["topology"][0]["node"]
 266             for node in nodes:
 267                 tpoints = node["termination-point"]
 268                 for tpoint in tpoints:
 269                     t_url = (
 270                         "http://"
 271                         + self.host
 272                         + ":"
 273                         + self.port
 274                         + "/restconf/operational/network-topology:network-topology/topology/flow:1/node/"
 275                         + node["node-id"]
 276                         + "/termination-point/"
 277                         + tpoint["tp-id"]
 278                     )
 279                     url_list.append(t_url)
 280             return url_list
 281         except KeyError:
 282             print("Error parsing topology json")
 283             return []
 284
 285
 286 class InvUrlGenerator(TestUrlGenerator):
 287     """
 288     Class to generate test URLs from the inventory shard.
 289     """
 290
 291     def __init__(self, host, port, auth):
 292         TestUrlGenerator.__init__(self, host, port, auth)
 293         self.resource_string = "restconf/operational/opendaylight-inventory:nodes"
 294
 295     def url_generator(self, inv_data):
 296         url_list = []
 297         try:
 298             nodes = inv_data["nodes"]["node"]
 299             for node in nodes:
 300                 nconns = node["node-connector"]
 301                 for nconn in nconns:
 302                     i_url = (
 303                         "http://"
 304                         + self.host
 305                         + ":"
 306                         + self.port
 307                         + "/restconf/operational/opendaylight-inventory:nodes/node/"
 308                         + node["id"]
 309                         + "/node-connector/"
 310                         + nconn["id"]
 311                         + "/opendaylight-port-statistics:flow-capable-node-connector-statistics"
 312                     )
 313                     url_list.append(i_url)
 314             return url_list
 315         except KeyError:
 316             print("Error parsing inventory json")
 317             return []
 318
 319
 320 if __name__ == "__main__":
 321     parser = argparse.ArgumentParser(
 322         description="Flow programming performance test: First adds and then deletes flows "
 323         "into the config tree, as specified by optional parameters."
 324     )
 325
 326     parser.add_argument(
 327         "--host",
 328         default="127.0.0.1",
 329         help="Host where odl controller is running (default is 127.0.0.1)",
 330     )
 331     parser.add_argument(
 332         "--port",
 333         default="8181",
 334         help="Port on which odl's RESTCONF is listening (default is 8181)",
 335     )
 336     parser.add_argument(
 337         "--auth",
 338         dest="auth",
 339         action="store_true",
 340         default=False,
 341         help="Use the ODL default username/password 'admin'/'admin' to authenticate access to REST; "
 342         "default: no authentication",
 343     )
 344     parser.add_argument(
 345         "--threads",
 346         type=int,
 347         default=1,
 348         help="Number of request worker threads to start in each cycle; default=1. ",
 349     )
 350     parser.add_argument(
 351         "--requests",
 352         type=int,
 353         default=100,
 354         help="Number of requests each worker thread will send to the controller; default=100.",
 355     )
 356     parser.add_argument(
 357         "--resource",
 358         choices=["inv", "topo", "topo+inv", "all"],
 359         default="both",
 360         help="Which resource to test: inventory, topology, or both; default both",
 361     )
 362     parser.add_argument(
 363         "--plevel",
 364         type=int,
 365         default=0,
 366         help="Print level: controls output verbosity. 0-lowest, 1-highest; default 0",
 367     )
 368     in_args = parser.parse_args()
 369
 370     topo_urls = []
 371     inv_urls = []
 372
 373     # If required, get topology resource URLs
 374     if in_args.resource != "inventory":
 375         tg = TopoUrlGenerator(in_args.host, in_args.port, in_args.auth)
 376         topo_urls += tg.generate()
 377         if len(topo_urls) == 0:
 378             print("Failed to generate topology URLs")
 379             sys.exit(-1)
 380
 381     # If required, get inventory resource URLs
 382     if in_args.resource != "topology":
 383         ig = InvUrlGenerator(in_args.host, in_args.port, in_args.auth)
 384         inv_urls += ig.generate()
 385         if len(inv_urls) == 0:
 386             print("Failed to generate inventory URLs")
 387             sys.exit(-1)
 388
 389     if in_args.resource == "topo+inv" or in_args.resource == "all":
 390         # To have balanced test results, the number of URLs for topology and inventory must be the same
 391         if len(topo_urls) != len(inv_urls):
 392             print("The number of topology and inventory URLs don't match")
 393             sys.exit(-1)
 394
 395     st = ShardPerformanceTester(
 396         in_args.host,
 397         in_args.port,
 398         in_args.auth,
 399         in_args.threads,
 400         in_args.requests,
 401         in_args.plevel,
 402     )
 403
 404     if in_args.resource == "all" or in_args.resource == "topo":
 405         print("===================================")
 406         print("Testing topology shard performance:")
 407         print("===================================")
 408         st.run_test(topo_urls)
 409
 410     if in_args.resource == "all" or in_args.resource == "inv":
 411         print("====================================")
 412         print("Testing inventory shard performance:")
 413         print("====================================")
 414         st.run_test(inv_urls)
 415
 416     if in_args.resource == "topo+inv" or in_args.resource == "all":
 417         print("===============================================")
 418         print("Testing combined shards (topo+inv) performance:")
 419         print("===============================================")
 420         st.run_test(topo_urls + inv_urls)