test/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 = {
  35     u'flow': [
  36         {
  37             u'hard-timeout': 65000,
  38             u'idle-timeout': 65000,
  39             u'cookie_mask': 4294967295,
  40             u'flow-name': u'FLOW-NAME-TEMPLATE',
  41             u'priority': 2,
  42             u'strict': False,
  43             u'cookie': 0,
  44             u'table_id': 0,
  45             u'installHw': False,
  46             u'id': u'FLOW-ID-TEMPLATE',
  47             u'match': {
  48                 u'ipv4-destination': u'0.0.0.0/32',
  49                 u'ethernet-match': {
  50                     u'ethernet-type': {
  51                         u'type': 2048
  52                     }
  53                 }
  54             },
  55             u'instructions': {
  56                 u'instruction': [
  57                     {
  58                         u'order': 0,
  59                         u'apply-actions': {
  60                             u'action': [
  61                                 {
  62                                     u'drop-action': {},
  63                                     u'order': 0
  64                                 }
  65                             ]
  66                         }
  67                     }
  68                 ]
  69             }
  70         }
  71     ]
  72 }
  73
  74
  75 def _get_notes(fldet=[]):
  76     '''For given list of flow details it produces a dictionary with statistics
  77     { swId1 : { tabId1 : flows_count1,
  78                 tabId2 : flows_count2,
  79                ...
  80                 'total' : switch count }
  81       swId2 ...
  82     }
  83     '''
  84     notes = {}
  85     for (sw, tab, flow) in fldet:
  86         if sw not in notes:
  87             notes[sw] = {'total': 0}
  88         if tab not in notes[sw]:
  89             notes[sw][tab] = 0
  90         notes[sw][tab] += 1
  91         notes[sw]['total'] += 1
  92     return notes
  93
  94
  95 def _randomize(spread, maxn):
  96     '''Returns a randomized switch or table id'''
  97     if spread not in _spreads:
  98         raise Exception('Spread method {} not available'.format(spread))
  99     while True:
 100         if spread == 'gauss':
 101             ga = abs(random.gauss(0, 1))
 102             rv = int(ga*float(maxn)/3)
 103             if rv < maxn:
 104                 return rv
 105         elif spread == 'linear':
 106             rv = int(random.random() * float(maxn))
 107             if rv < maxn:
 108                 return rv
 109             else:
 110                 raise ValueError('rv >= maxn')
 111         elif spread == 'first':
 112             return 0
 113
 114
 115 def generate_new_flow_details(flows=10, switches=1, swspread='gauss', tables=250, tabspread='gauss'):
 116     """Generate a list of tupples (switch_id, table_id, flow_id) which are generated
 117     according to the spread rules between swithces and tables.
 118     It also returns a dictionary with statsistics."""
 119     swflows = [_randomize(swspread, switches) for f in range(int(flows))]
 120     fltables = [(s, _randomize(tabspread, tables), idx) for idx, s in enumerate(swflows)]
 121     notes = _get_notes(fltables)
 122     return fltables, notes
 123
 124
 125 def _prepare_add(cntl, sw, tab, fl, ip, template=None):
 126     '''Creates a PUT http requests to configure a flow in configuration datastore'''
 127     url = 'http://'+cntl+':'+'8181'
 128     url += '/restconf/config/opendaylight-inventory:nodes/node/openflow:'+str(sw)+'/table/'+str(tab)+'/flow/'+str(fl)
 129     flow = copy.deepcopy(template['flow'][0])
 130     flow['cookie'] = fl
 131     flow['flow-name'] = 'TestFlow-%d' % fl
 132     flow['id'] = str(fl)
 133     flow['match']['ipv4-destination'] = '%s/32' % str(netaddr.IPAddress(ip))
 134     flow['table_id'] = tab
 135     fmod = dict(template)
 136     fmod['flow'] = flow
 137     req_data = json.dumps(fmod)
 138     req = requests.Request('PUT', url, headers={'Content-Type': 'application/json'}, data=req_data,
 139                            auth=('admin', 'admin'))
 140     return req
 141
 142
 143 def _prepare_table_add(cntl, flows, template=None):
 144     '''Creates a POST http requests to configure a flow in configuration datastore'''
 145     f1 = flows[0]
 146     sw, tab, fl, ip = f1
 147     url = 'http://'+cntl+':'+'8181'
 148     url += '/restconf/config/opendaylight-inventory:nodes/node/openflow:'+str(sw)+'/table/'+str(tab)
 149     fdets = []
 150     for sw, tab, fl, ip in flows:
 151         flow = copy.deepcopy(template['flow'][0])
 152         flow['cookie'] = fl
 153         flow['flow-name'] = 'TestFlow-%d' % fl
 154         flow['id'] = str(fl)
 155         flow['match']['ipv4-destination'] = '%s/32' % str(netaddr.IPAddress(ip))
 156         flow['table_id'] = tab
 157         fdets.append(flow)
 158     fmod = dict(template)
 159     fmod['flow'] = fdets
 160     req_data = json.dumps(fmod)
 161     req = requests.Request('POST', url, headers={'Content-Type': 'application/json'}, data=req_data,
 162                            auth=('admin', 'admin'))
 163     return req
 164
 165
 166 def _prepare_delete(cntl, sw, tab, fl, ip, template=None):
 167     '''Creates a DELETE http request to remove the flow from configuration datastore'''
 168     url = 'http://'+cntl+':'+'8181'
 169     url += '/restconf/config/opendaylight-inventory:nodes/node/openflow:'+str(sw)+'/table/'+str(tab)+'/flow/'+str(fl)
 170     req = requests.Request('DELETE', url, headers={'Content-Type': 'application/json'}, auth=('admin', 'admin'))
 171     return req
 172
 173
 174 def _wt_request_sender(thread_id, preparefnc, inqueue=None, exitevent=None, controllers=[], restport='', template=None,
 175                        outqueue=None):
 176     '''The funcion runs in a thread. It reads out flow details from the queue and configures
 177     the flow on the controller'''
 178     ses = requests.Session()
 179     cntl = controllers[0]
 180     counter = [0 for i in range(600)]
 181
 182     while True:
 183         try:
 184             (sw, tab, fl, ip) = inqueue.get(timeout=1)
 185             sw, tab, fl, ip = sw+1, tab, fl+1, ip
 186         except Queue.Empty:
 187             if exitevent.is_set() and inqueue.empty():
 188                 break
 189             continue
 190         req = preparefnc(cntl, sw, tab, fl, ip, template=template)
 191         prep = ses.prepare_request(req)
 192         try:
 193             rsp = ses.send(prep, timeout=5)
 194         except requests.exceptions.Timeout:
 195             counter[99] += 1
 196             continue
 197         counter[rsp.status_code] += 1
 198     res = {}
 199     for i, v in enumerate(counter):
 200         if v > 0:
 201             res[i] = v
 202     outqueue.put(res)
 203
 204
 205 def _wt_bulk_request_sender(thread_id, preparefnc, inqueue=None, exitevent=None, controllers=[], restport='',
 206                             template=None, outqueue=None):
 207     '''The funcion runs in a thread. It reads out flow details from the queue and configures
 208     the flow on the controller'''
 209     ses = requests.Session()
 210     cntl = controllers[0]
 211     counter = [0 for i in range(600)]
 212     loop = True
 213
 214     while loop:
 215         try:
 216             flowlist = inqueue.get(timeout=1)
 217         except Queue.Empty:
 218             if exitevent.is_set() and inqueue.empty():
 219                 loop = False
 220             continue
 221         req = preparefnc(cntl, flowlist, template=template)
 222         prep = ses.prepare_request(req)
 223         try:
 224             rsp = ses.send(prep, timeout=5)
 225         except requests.exceptions.Timeout:
 226             counter[99] += 1
 227             continue
 228         counter[rsp.status_code] += 1
 229     res = {}
 230     for i, v in enumerate(counter):
 231         if v > 0:
 232             res[i] = v
 233     outqueue.put(res)
 234
 235
 236 def _config_task_executor(preparefnc, flow_details=[], flow_template=None, controllers=['127.0.0.1'], restport='8181',
 237                           nrthreads=1):
 238     '''Function starts thread executors and put required information to the queue. Executors read the queue and send
 239     http requests. After the thread's join, it produces a summary result.'''
 240     # TODO: multi controllers support
 241     ip_addr = Counter(int(netaddr.IPAddress('10.0.0.1')))
 242     if flow_template is not None:
 243         template = flow_template
 244     else:
 245         template = _default_flow_template_json
 246
 247     # lets enlarge the tupple of flow details with IP, to be used with the template
 248     flows = [(s, t, f, ip_addr.increment()) for s, t, f in flow_details]
 249
 250     # lets fill the qurue
 251     q = Queue.Queue()
 252     for f in flows:
 253         q.put(f)
 254
 255     # result_gueue
 256     rq = Queue.Queue()
 257     # creaet exit event
 258     ee = threading.Event()
 259
 260     # lets start threads whic will read flow details fro queues and send
 261     threads = []
 262     for i in range(int(nrthreads)):
 263         t = threading.Thread(target=_wt_request_sender, args=(i, preparefnc),
 264                              kwargs={"inqueue": q, "exitevent": ee, "controllers": controllers, "restport": restport,
 265                                      "template": template, "outqueue": rq})
 266         threads.append(t)
 267         t.start()
 268
 269     ee.set()
 270
 271     result = {}
 272     # waitng for them
 273     for t in threads:
 274         t.join()
 275         res = rq.get()
 276         for k, v in res.iteritems():
 277             if k not in result:
 278                 result[k] = v
 279             else:
 280                 result[k] += v
 281     return result
 282
 283
 284 def configure_flows(*args, **kwargs):
 285     '''Configure flows based on given flow details
 286     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 287                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 288     return _config_task_executor(_prepare_add, *args, **kwargs)
 289
 290
 291 def deconfigure_flows(*args, **kwargs):
 292     '''Deconfigure flows based on given flow details.
 293     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 294                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 295     return _config_task_executor(_prepare_delete, *args, **kwargs)
 296
 297
 298 def _bulk_config_task_executor(preparefnc, flow_details=[], flow_template=None, controllers=['127.0.0.1'],
 299                                restport='8181', nrthreads=1, fpr=1):
 300     '''Function starts thread executors and put required information to the queue. Executors read the queue and send
 301     http requests. After the thread's join, it produces a summary result.'''
 302     # TODO: multi controllers support
 303     ip_addr = Counter(int(netaddr.IPAddress('10.0.0.1')))
 304     if flow_template is not None:
 305         template = flow_template
 306     else:
 307         template = _default_flow_template_json
 308
 309     # lets enlarge the tupple of flow details with IP, to be used with the template
 310     flows = [(s, t, f, ip_addr.increment()) for s, t, f in flow_details]
 311     # lest divide flows into switches and tables
 312     fg = {}
 313     for fl in flows:
 314         s, t, f, ip = fl
 315         fk = (s, t)
 316         if (s, t) in fg:
 317             fg[fk].append(fl)
 318         else:
 319             fg[fk] = [fl]
 320
 321     # lets fill the qurue
 322     q = Queue.Queue()
 323     for k, v in fg.iteritems():
 324         while len(v) > 0:
 325             q.put(v[:int(fpr)])
 326             v = v[int(fpr):]
 327
 328     # result_gueue
 329     rq = Queue.Queue()
 330     # creaet exit event
 331     ee = threading.Event()
 332
 333     # lets start threads whic will read flow details fro queues and send
 334     threads = []
 335     for i in range(int(nrthreads)):
 336         t = threading.Thread(target=_wt_bulk_request_sender, args=(i, preparefnc),
 337                              kwargs={"inqueue": q, "exitevent": ee, "controllers": controllers, "restport": restport,
 338                                      "template": template, "outqueue": rq})
 339         threads.append(t)
 340         t.start()
 341
 342     ee.set()
 343
 344     result = {}
 345     # waitng for them
 346     for t in threads:
 347         t.join()
 348         res = rq.get()
 349         for k, v in res.iteritems():
 350             if k not in result:
 351                 result[k] = v
 352             else:
 353                 result[k] += v
 354     return result
 355
 356
 357 def configure_flows_bulk(*args, **kwargs):
 358     '''Configure flows based on given flow details
 359     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 360                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 361     return _bulk_config_task_executor(_prepare_table_add, *args, **kwargs)
 362
 363
 364 def _get_operational_inventory_of_switches(controller):
 365     '''GET requests to get operational inventory node details'''
 366     url = 'http://'+controller+':8181/restconf/operational/opendaylight-inventory:nodes'
 367     rsp = requests.get(url, headers={'Accept': 'application/json'}, stream=False, auth=('admin', 'admin'))
 368     if rsp.status_code != 200:
 369         return None
 370     inv = json.loads(rsp.content)
 371     if 'nodes' not in inv:
 372         return None
 373     if 'node' not in inv['nodes']:
 374         return []
 375     inv = inv['nodes']['node']
 376     switches = [sw for sw in inv if 'openflow:' in sw['id']]
 377     return switches
 378
 379
 380 def flow_stats_collected(controller=''):
 381     '''Once flows are configured, thisfunction is used to check if flows are present in the operational datastore'''
 382     # print type(flow_details), flow_details
 383     active_flows = 0
 384     found_flows = 0
 385     switches = _get_operational_inventory_of_switches(controller)
 386     if switches is None:
 387         return 0, 0, 0
 388     for sw in switches:
 389         tabs = sw['flow-node-inventory:table']
 390         for t in tabs:
 391             active_flows += t['opendaylight-flow-table-statistics:flow-table-statistics']['active-flows']
 392             if 'flow' in t:
 393                 found_flows += len(t['flow'])
 394     print "Switches,ActiveFlows(reported)/FlowsFound", len(switches), active_flows, found_flows
 395     return len(switches), active_flows, found_flows
 396
 397
 398 def get_switches_count(controller=''):
 399     '''Count the switches presnt in the operational inventory nodes datastore'''
 400     switches = _get_operational_inventory_of_switches(controller)
 401     if switches is None:
 402         return 0
 403     return len(switches)