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         prep = req.prepare()
 193         try:
 194             rsp = ses.send(prep, timeout=5)
 195         except requests.exceptions.Timeout:
 196             counter[99] += 1
 197             continue
 198         counter[rsp.status_code] += 1
 199     res = {}
 200     for i, v in enumerate(counter):
 201         if v > 0:
 202             res[i] = v
 203     outqueue.put(res)
 204
 205
 206 def _wt_bulk_request_sender(thread_id, preparefnc, inqueue=None, exitevent=None, controllers=[], restport='',
 207                             template=None, outqueue=None):
 208     '''The funcion runs in a thread. It reads out flow details from the queue and configures
 209     the flow on the controller'''
 210     ses = requests.Session()
 211     cntl = controllers[0]
 212     counter = [0 for i in range(600)]
 213     loop = True
 214
 215     while loop:
 216         try:
 217             flowlist = inqueue.get(timeout=1)
 218         except Queue.Empty:
 219             if exitevent.is_set() and inqueue.empty():
 220                 loop = False
 221             continue
 222         req = preparefnc(cntl, flowlist, template=template)
 223         # prep = ses.prepare_request(req)
 224         prep = req.prepare()
 225         try:
 226             rsp = ses.send(prep, timeout=5)
 227         except requests.exceptions.Timeout:
 228             counter[99] += 1
 229             continue
 230         counter[rsp.status_code] += 1
 231     res = {}
 232     for i, v in enumerate(counter):
 233         if v > 0:
 234             res[i] = v
 235     outqueue.put(res)
 236
 237
 238 def _config_task_executor(preparefnc, flow_details=[], flow_template=None, controllers=['127.0.0.1'], restport='8181',
 239                           nrthreads=1):
 240     '''Function starts thread executors and put required information to the queue. Executors read the queue and send
 241     http requests. After the thread's join, it produces a summary result.'''
 242     # TODO: multi controllers support
 243     ip_addr = Counter(int(netaddr.IPAddress('10.0.0.1')))
 244     if flow_template is not None:
 245         template = flow_template
 246     else:
 247         template = _default_flow_template_json
 248
 249     # lets enlarge the tupple of flow details with IP, to be used with the template
 250     flows = [(s, t, f, ip_addr.increment()) for s, t, f in flow_details]
 251
 252     # lets fill the qurue
 253     q = Queue.Queue()
 254     for f in flows:
 255         q.put(f)
 256
 257     # result_gueue
 258     rq = Queue.Queue()
 259     # creaet exit event
 260     ee = threading.Event()
 261
 262     # lets start threads whic will read flow details fro queues and send
 263     threads = []
 264     for i in range(int(nrthreads)):
 265         t = threading.Thread(target=_wt_request_sender, args=(i, preparefnc),
 266                              kwargs={"inqueue": q, "exitevent": ee, "controllers": controllers, "restport": restport,
 267                                      "template": template, "outqueue": rq})
 268         threads.append(t)
 269         t.start()
 270
 271     ee.set()
 272
 273     result = {}
 274     # waitng for them
 275     for t in threads:
 276         t.join()
 277         res = rq.get()
 278         for k, v in res.iteritems():
 279             if k not in result:
 280                 result[k] = v
 281             else:
 282                 result[k] += v
 283     return result
 284
 285
 286 def configure_flows(*args, **kwargs):
 287     '''Configure flows based on given flow details
 288     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 289                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 290     return _config_task_executor(_prepare_add, *args, **kwargs)
 291
 292
 293 def deconfigure_flows(*args, **kwargs):
 294     '''Deconfigure flows based on given flow details.
 295     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 296                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 297     return _config_task_executor(_prepare_delete, *args, **kwargs)
 298
 299
 300 def _bulk_config_task_executor(preparefnc, flow_details=[], flow_template=None, controllers=['127.0.0.1'],
 301                                restport='8181', nrthreads=1, fpr=1):
 302     '''Function starts thread executors and put required information to the queue. Executors read the queue and send
 303     http requests. After the thread's join, it produces a summary result.'''
 304     # TODO: multi controllers support
 305     ip_addr = Counter(int(netaddr.IPAddress('10.0.0.1')))
 306     if flow_template is not None:
 307         template = flow_template
 308     else:
 309         template = _default_flow_template_json
 310
 311     # lets enlarge the tupple of flow details with IP, to be used with the template
 312     flows = [(s, t, f, ip_addr.increment()) for s, t, f in flow_details]
 313     # lest divide flows into switches and tables
 314     fg = {}
 315     for fl in flows:
 316         s, t, f, ip = fl
 317         fk = (s, t)
 318         if (s, t) in fg:
 319             fg[fk].append(fl)
 320         else:
 321             fg[fk] = [fl]
 322
 323     # lets fill the qurue
 324     q = Queue.Queue()
 325     for k, v in fg.iteritems():
 326         while len(v) > 0:
 327             q.put(v[:int(fpr)])
 328             v = v[int(fpr):]
 329
 330     # result_gueue
 331     rq = Queue.Queue()
 332     # creaet exit event
 333     ee = threading.Event()
 334
 335     # lets start threads whic will read flow details fro queues and send
 336     threads = []
 337     for i in range(int(nrthreads)):
 338         t = threading.Thread(target=_wt_bulk_request_sender, args=(i, preparefnc),
 339                              kwargs={"inqueue": q, "exitevent": ee, "controllers": controllers, "restport": restport,
 340                                      "template": template, "outqueue": rq})
 341         threads.append(t)
 342         t.start()
 343
 344     ee.set()
 345
 346     result = {}
 347     # waitng for them
 348     for t in threads:
 349         t.join()
 350         res = rq.get()
 351         for k, v in res.iteritems():
 352             if k not in result:
 353                 result[k] = v
 354             else:
 355                 result[k] += v
 356     return result
 357
 358
 359 def configure_flows_bulk(*args, **kwargs):
 360     '''Configure flows based on given flow details
 361     Input parameters with default values: preparefnc, flow_details=[], flow_template=None,
 362                                controllers=['127.0.0.1'], restport='8181', nrthreads=1'''
 363     return _bulk_config_task_executor(_prepare_table_add, *args, **kwargs)
 364
 365
 366 def _get_operational_inventory_of_switches(controller):
 367     '''GET requests to get operational inventory node details'''
 368     url = 'http://'+controller+':8181/restconf/operational/opendaylight-inventory:nodes'
 369     rsp = requests.get(url, headers={'Accept': 'application/json'}, stream=False, auth=('admin', 'admin'))
 370     if rsp.status_code != 200:
 371         return None
 372     inv = json.loads(rsp.content)
 373     if 'nodes' not in inv:
 374         return None
 375     if 'node' not in inv['nodes']:
 376         return []
 377     inv = inv['nodes']['node']
 378     switches = [sw for sw in inv if 'openflow:' in sw['id']]
 379     return switches
 380
 381
 382 def flow_stats_collected(controller=''):
 383     '''Once flows are configured, thisfunction is used to check if flows are present in the operational datastore'''
 384     # print type(flow_details), flow_details
 385     active_flows = 0
 386     found_flows = 0
 387     switches = _get_operational_inventory_of_switches(controller)
 388     if switches is None:
 389         return 0, 0, 0
 390     for sw in switches:
 391         tabs = sw['flow-node-inventory:table']
 392         for t in tabs:
 393             active_flows += t['opendaylight-flow-table-statistics:flow-table-statistics']['active-flows']
 394             if 'flow' in t:
 395                 found_flows += len(t['flow'])
 396     print "Switches,ActiveFlows(reported)/FlowsFound", len(switches), active_flows, found_flows
 397     return len(switches), active_flows, found_flows
 398
 399
 400 def get_switches_count(controller=''):
 401     '''Count the switches presnt in the operational inventory nodes datastore'''
 402     switches = _get_operational_inventory_of_switches(controller)
 403     if switches is None:
 404         return 0
 405     return len(switches)