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     headers = {'Accept': 'application/json'}
  58
  59     def __init__(self, host, port, auth, threads, nrequests, plevel):
  60         """
  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(r_url, headers=self.headers, stream=False, auth=('admin', 'admin'))
  92         return r.status_code
  93
  94     def worker(self, tid, urls):
  95         """
  96         Worker thread function. Connects to system-under-test and makes 'self.requests' requests for
  97         resources to URLs randomly selected from 'urls'
  98         :param tid: Worker thread ID
  99         :param urls: List of resource URLs
 100         :return: None
 101         """
 102         res = {200: 0}
 103
 104         s = requests.Session()
 105
 106         with self.print_lock:
 107             print '    Thread %d: Performing %d requests' % (tid, self.requests)
 108
 109         with Timer() as t:
 110             for r in range(self.requests):
 111                 sts = self.make_request(s, urls)
 112                 try:
 113                     res[sts] += 1
 114                 except KeyError:
 115                     res[sts] = 1
 116
 117         ok_rate = res[200] / t.secs
 118         total_rate = sum(res.values()) / t.secs
 119
 120         with self.print_lock:
 121             print 'Thread %d done:' % tid
 122             print '    Time: %.2f,' % t.secs
 123             print '    Success rate:  %.2f, Total rate: %.2f' % (ok_rate, total_rate)
 124             print '    Per-thread stats: ',
 125             print res
 126             self.threads_done += 1
 127             self.total_rate += total_rate
 128
 129         s.close()
 130
 131         with self.cond:
 132             self.cond.notifyAll()
 133
 134     def run_test(self, urls):
 135         """
 136         Runs the performance test. Starts 'self.threads' worker threads, waits for all of them to finish and
 137         prints results.
 138         :param urls: List of urls from which to request resources
 139         :return: None
 140         """
 141
 142         threads = []
 143         self.total_rate = 0
 144
 145         # Initialize url counters
 146         del self.url_counters[:]
 147         for i in range(len(urls)):
 148             self.url_counters.append(Counter(0))
 149
 150         # Start all worker threads
 151         for i in range(self.threads):
 152             t = threading.Thread(target=self.worker, args=(i, urls))
 153             threads.append(t)
 154             t.start()
 155
 156         # Wait for all threads to finish and measure the execution time
 157         with Timer() as t:
 158             while self.threads_done < self.threads:
 159                 with self.cond:
 160                     self.cond.wait()
 161
 162         # Print summary results. Each worker prints its owns results too.
 163         print '\nSummary Results:'
 164         print '    Requests/sec (total_sum): %.2f' % ((self.threads * self.requests) / t.secs)
 165         print '    Requests/sec (measured):  %.2f' % ((self.threads * self.requests) / t.secs)
 166         print '    Time: %.2f' % t.secs
 167         self.threads_done = 0
 168
 169         if self.plevel > 0:
 170             print '    Per URL Counts: ',
 171             for i in range(len(urls)):
 172                 print '%d' % self.url_counters[i].value,
 173             print '\n'
 174
 175
 176 class TestUrlGenerator(object):
 177     """
 178     Base abstract class to generate test URLs for ShardPerformanceTester. First, an entire subtree representing
 179     a shard or a set of resources is retrieved, then a set of URLS to access small data stanzas is created. This
 180     class only defines the framework, the methods that create URL sets are defined in derived classes.
 181     """
 182
 183     def __init__(self, host, port, auth):
 184         """
 185         Initialization
 186         :param host: Controller's IP address
 187         :param port: Controller's RESTCONF port
 188         :param auth: Indicates whether to use authentication with default user/password (admin/admin)
 189         :return: None
 190         """
 191         self.host = host
 192         self.port = port
 193         self.auth = auth
 194         self.resource_string = ''
 195
 196     def url_generator(self, data):
 197         """
 198         Abstract  URL generator. Must be overridden in a derived class
 199         :param data: Bulk resource data (JSON) from which to generate the URLs
 200         :return: List of generated Resources
 201         """
 202         print "Abstract class '%s' should never be used standalone" % self.__class__.__name__
 203         return []
 204
 205     def generate(self):
 206         """
 207         Drives the generation of test URLs. First, it gets a 'bulk' resource (e.g. the entire inventory
 208          or the entire topology) from the controller specified during int()  and then invokes a resource-specific
 209          URL generator to create a set of resource-specific URLs.
 210         """
 211         t_url = 'http://' + self.host + ":" + self.port + '/' + self.resource_string
 212         headers = {'Accept': 'application/json'}
 213         r_url = []
 214
 215         if not self.auth:
 216             r = requests.get(t_url, headers=headers, stream=False)
 217         else:
 218             r = requests.get(t_url, headers=headers, stream=False, auth=('admin', 'admin'))
 219
 220         if r.status_code != 200:
 221             print "Failed to get HTTP response from '%s', code %d" % (t_url, r.status_code)
 222         else:
 223             try:
 224                 r_url = self.url_generator(json.loads(r.content))
 225             except:
 226                 print "Failed to get json from '%s'. Please make sure you are connected to mininet." % r_url
 227
 228         return r_url
 229
 230
 231 class TopoUrlGenerator(TestUrlGenerator):
 232     """
 233     Class to generate test URLs from the topology shard.
 234     :return: List of generated Resources
 235     """
 236     def __init__(self, host, port, auth):
 237         TestUrlGenerator.__init__(self, host, port, auth)
 238         self.resource_string = 'restconf/operational/network-topology:network-topology/topology/flow:1'
 239
 240     def url_generator(self, topo_data):
 241         url_list = []
 242         try:
 243             nodes = topo_data['topology'][0]['node']
 244             for node in nodes:
 245                 tpoints = node['termination-point']
 246                 for tpoint in tpoints:
 247                     t_url = 'http://' + self.host + ":" + self.port + \
 248                             '/restconf/operational/network-topology:network-topology/topology/flow:1/node/' + \
 249                             node['node-id'] + '/termination-point/' + tpoint['tp-id']
 250                     url_list.append(t_url)
 251             return url_list
 252         except KeyError:
 253             print 'Error parsing topology json'
 254             return []
 255
 256
 257 class InvUrlGenerator(TestUrlGenerator):
 258     """
 259     Class to generate test URLs from the inventory shard.
 260     """
 261
 262     def __init__(self, host, port, auth):
 263         TestUrlGenerator.__init__(self, host, port, auth)
 264         self.resource_string = 'restconf/operational/opendaylight-inventory:nodes'
 265
 266     def url_generator(self, inv_data):
 267         url_list = []
 268         try:
 269             nodes = inv_data['nodes']['node']
 270             for node in nodes:
 271                 nconns = node['node-connector']
 272                 for nconn in nconns:
 273                     i_url = 'http://' + self.host + ":" + self.port + \
 274                             '/restconf/operational/opendaylight-inventory:nodes/node/' + \
 275                             node['id'] + '/node-connector/' + nconn['id'] + \
 276                             '/opendaylight-port-statistics:flow-capable-node-connector-statistics'
 277                     url_list.append(i_url)
 278             return url_list
 279         except KeyError:
 280             print 'Error parsing inventory json'
 281             return []
 282
 283
 284 if __name__ == "__main__":
 285     parser = argparse.ArgumentParser(description='Flow programming performance test: First adds and then deletes flows '
 286                                                  'into the config tree, as specified by optional parameters.')
 287
 288     parser.add_argument('--host', default='127.0.0.1',
 289                         help='Host where odl controller is running (default is 127.0.0.1)')
 290     parser.add_argument('--port', default='8181',
 291                         help='Port on which odl\'s RESTCONF is listening (default is 8181)')
 292     parser.add_argument('--auth', dest='auth', action='store_true', default=False,
 293                         help="Use the ODL default username/password 'admin'/'admin' to authenticate access to REST; "
 294                              'default: no authentication')
 295     parser.add_argument('--threads', type=int, default=1,
 296                         help='Number of request worker threads to start in each cycle; default=1. ')
 297     parser.add_argument('--requests', type=int, default=100,
 298                         help='Number of requests each worker thread will send to the controller; default=100.')
 299     parser.add_argument('--resource', choices=['inv', 'topo', 'topo+inv', 'all'], default='both',
 300                         help='Which resource to test: inventory, topology, or both; default both')
 301     parser.add_argument('--plevel', type=int, default=0,
 302                         help='Print level: controls output verbosity. 0-lowest, 1-highest; default 0')
 303     in_args = parser.parse_args()
 304
 305     topo_urls = []
 306     inv_urls = []
 307
 308     # If required, get topology resource URLs
 309     if in_args.resource != 'inventory':
 310         tg = TopoUrlGenerator(in_args.host, in_args.port, in_args.auth)
 311         topo_urls += tg.generate()
 312         if len(topo_urls) == 0:
 313             print 'Failed to generate topology URLs'
 314             sys.exit(-1)
 315
 316     # If required, get inventory resource URLs
 317     if in_args.resource != 'topology':
 318         ig = InvUrlGenerator(in_args.host, in_args.port, in_args.auth)
 319         inv_urls += ig.generate()
 320         if len(inv_urls) == 0:
 321             print 'Failed to generate inventory URLs'
 322             sys.exit(-1)
 323
 324     if in_args.resource == 'topo+inv' or in_args.resource == 'all':
 325         # To have balanced test results, the number of URLs for topology and inventory must be the same
 326         if len(topo_urls) != len(inv_urls):
 327             print "The number of topology and inventory URLs don't match"
 328             sys.exit(-1)
 329
 330     st = ShardPerformanceTester(in_args.host, in_args.port, in_args.auth, in_args.threads, in_args.requests,
 331                                 in_args.plevel)
 332
 333     if in_args.resource == 'all' or in_args.resource == 'topo':
 334         print '==================================='
 335         print 'Testing topology shard performance:'
 336         print '==================================='
 337         st.run_test(topo_urls)
 338
 339     if in_args.resource == 'all' or in_args.resource == 'inv':
 340         print '===================================='
 341         print 'Testing inventory shard performance:'
 342         print '===================================='
 343         st.run_test(inv_urls)
 344
 345     if in_args.resource == 'topo+inv' or in_args.resource == 'all':
 346         print '==============================================='
 347         print 'Testing combined shards (topo+inv) performance:'
 348         print '==============================================='
 349         st.run_test(topo_urls + inv_urls)