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

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