2 The purpose of this library is the ability to spread configured flows
3 over the specified tables and switches.
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
18 class Counter(object):
19 def __init__(self, start=0):
20 self.lock = threading.Lock()
23 def increment(self, value=1):
33 _spreads = ['gauss', 'linear', 'first'] # possible defined spreads at the moment
34 _default_flow_template_json = {
37 u'hard-timeout': 65000,
38 u'idle-timeout': 65000,
39 u'cookie_mask': 4294967295,
40 u'flow-name': u'FLOW-NAME-TEMPLATE',
46 u'id': u'FLOW-ID-TEMPLATE',
48 u'ipv4-destination': u'0.0.0.0/32',
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,
80 'total' : switch count }
85 for (sw, tab, flow) in fldet:
87 notes[sw] = {'total': 0}
88 if tab not in notes[sw]:
91 notes[sw]['total'] += 1
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))
100 if spread == 'gauss':
101 ga = abs(random.gauss(0, 1))
102 rv = int(ga*float(maxn)/3)
105 elif spread == 'linear':
106 rv = int(random.random() * float(maxn))
110 raise ValueError('rv >= maxn')
111 elif spread == 'first':
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
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])
131 flow['flow-name'] = 'TestFlow-%d' % fl
133 flow['match']['ipv4-destination'] = '%s/32' % str(netaddr.IPAddress(ip))
134 flow['table_id'] = tab
135 fmod = dict(template)
137 req_data = json.dumps(fmod)
138 req = requests.Request('PUT', url, headers={'Content-Type': 'application/json'}, data=req_data,
139 auth=('admin', 'admin'))
143 def _prepare_table_add(cntl, flows, template=None):
144 '''Creates a POST http requests to configure a flow in configuration datastore'''
147 url = 'http://'+cntl+':'+'8181'
148 url += '/restconf/config/opendaylight-inventory:nodes/node/openflow:'+str(sw)+'/table/'+str(tab)
150 for sw, tab, fl, ip in flows:
151 flow = copy.deepcopy(template['flow'][0])
153 flow['flow-name'] = 'TestFlow-%d' % fl
155 flow['match']['ipv4-destination'] = '%s/32' % str(netaddr.IPAddress(ip))
156 flow['table_id'] = tab
158 fmod = dict(template)
160 req_data = json.dumps(fmod)
161 req = requests.Request('POST', url, headers={'Content-Type': 'application/json'}, data=req_data,
162 auth=('admin', 'admin'))
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'))
174 def _wt_request_sender(thread_id, preparefnc, inqueue=None, exitevent=None, controllers=[], restport='', template=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)]
184 (sw, tab, fl, ip) = inqueue.get(timeout=1)
185 sw, tab, fl, ip = sw+1, tab, fl+1, ip
187 if exitevent.is_set() and inqueue.empty():
190 req = preparefnc(cntl, sw, tab, fl, ip, template=template)
191 prep = ses.prepare_request(req)
193 rsp = ses.send(prep, timeout=5)
194 except requests.exceptions.Timeout:
197 counter[rsp.status_code] += 1
199 for i, v in enumerate(counter):
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)]
216 flowlist = inqueue.get(timeout=1)
218 if exitevent.is_set() and inqueue.empty():
221 req = preparefnc(cntl, flowlist, template=template)
222 prep = ses.prepare_request(req)
224 rsp = ses.send(prep, timeout=5)
225 except requests.exceptions.Timeout:
228 counter[rsp.status_code] += 1
230 for i, v in enumerate(counter):
236 def _config_task_executor(preparefnc, flow_details=[], flow_template=None, controllers=['127.0.0.1'], restport='8181',
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
245 template = _default_flow_template_json
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]
250 # lets fill the qurue
258 ee = threading.Event()
260 # lets start threads whic will read flow details fro queues and send
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})
276 for k, v in res.iteritems():
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)
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)
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
307 template = _default_flow_template_json
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
321 # lets fill the qurue
323 for k, v in fg.iteritems():
331 ee = threading.Event()
333 # lets start threads whic will read flow details fro queues and send
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})
349 for k, v in res.iteritems():
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)
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:
370 inv = json.loads(rsp.content)
371 if 'nodes' not in inv:
373 if 'node' not in inv['nodes']:
375 inv = inv['nodes']['node']
376 switches = [sw for sw in inv if 'openflow:' in sw['id']]
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
385 switches = _get_operational_inventory_of_switches(controller)
389 tabs = sw['flow-node-inventory:table']
391 active_flows += t['opendaylight-flow-table-statistics:flow-table-statistics']['active-flows']
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
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)