-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy pathFlowMonitor_3.py
473 lines (349 loc) · 18.8 KB
/
FlowMonitor_3.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
""" Module that monitors the average network interface occupation """
import subprocess
from collections import deque
import threading
import application_switch_3
import SwitchProperties
import time
class FlowMonitor_3:
""" Class that monitors network interface occupation """
def __init__(self, samples=10, period=3, interval_time=1.0, upper_limit=10*0.8, lower_limit=10*0.6):
self.n_samples = samples
self.period = period
self.interval_time = interval_time
self.switch_properties = SwitchProperties.SwitchProperties()
self.interfaces_list = self.switch_properties.get_interfaces()
self.complete_interface_list = []
self.old_queue_list = []
self.queues_ids = []
self.qos_register = dict.fromkeys(['uuid','port', 'id', 'min-rate', 'max-rate'] )
self.lock = threading.Lock()
for i in range(len(self.interfaces_list)):
complete_interface_dict = dict.fromkeys(['name', 'dpid', 'capacity', 'lower_limit', 'upper_limit', 'threshold', 'samples', 'use_averages', 'monitoring', 'is_congested', 'queueList'])
complete_interface_dict['name'] = self.interfaces_list[i]['name']
complete_interface_dict['dpid'] = self.interfaces_list[i]['dpid']
complete_interface_dict['capacity'] = self.interfaces_list[i]['capacity']
complete_interface_dict['lower_limit'] = lower_limit
complete_interface_dict['upper_limit'] = upper_limit
complete_interface_dict['threshold'] = upper_limit
complete_interface_dict['samples'] = []
complete_interface_dict['prevEma'] = 0
complete_interface_dict['currentEma'] = 0
complete_interface_dict['use_averages'] = 0
complete_interface_dict['monitoring'] = 0
complete_interface_dict['is_congested'] = 0
complete_interface_dict['queueList'] = []
self.complete_interface_list.append(complete_interface_dict)
for i in range(len(self.complete_interface_list)):
self.complete_interface_list[i]['use_averages'] = deque( maxlen=self.n_samples )
#Control variables
self.threads_id = []
self.init_window()
def reset_queues(self):
""" Clears QoS queues in all interfaces """
for i in range(len(self.complete_interface_list)):
subprocess.check_output('ovs-ofctl del-flows ' + self.complete_interface_list[i]['name'], shell=True)
subprocess.check_output('./clear_queues.sh ', shell=True)
def init_window(self):
""" Inits samples window """
for j in range(len(self.complete_interface_list)):
for i in range(self.n_samples):
self.complete_interface_list[j]['use_averages'].append(0)
for i in range(self.n_samples):
#sample list of dicts, each dict has ['name']['sample']
result = self.get_sample()
for j in range(len(self.complete_interface_list)):
last_samples = result[j]['sample']
self.complete_interface_list[j]['use_averages'].popleft()
self.complete_interface_list[j]['use_averages'].append(last_samples)
if i == 0:
self.complete_interface_list[j]['prevema'] = last_samples
for j in range(len(self.complete_interface_list)):
for a_bar in enumerate(self.complete_interface_list[j]['use_averages']):
self.complete_interface_list[j]['currentEma'] = self.ema(a_bar, self.complete_interface_list[j]['use_averages'], self.period, self.complete_interface_list[j]['prevEma'], smoothing=None)
self.complete_interface_list[j]['prevEma'] = self.complete_interface_list[j]['currentEma']
def update_window(self):
""" Updates the sample window """
for i in range(self.n_samples):
# Sample list of dicts, each dict has ['name']['sample']
result = self.get_sample() # < ---- GOTTA CHECK THIS
last_samples=0
for j in range(len(self.complete_interface_list)):
last_samples = result[j]['sample']
self.complete_interface_list[j]['use_averages'].popleft()
self.complete_interface_list[j]['use_averages'].append(last_samples)
for j in range(len(self.complete_interface_list)):
if i == 0:
self.complete_interface_list[j]['prevema'] = last_samples
for a_bar in enumerate(self.complete_interface_list[j]['use_averages']):
self.complete_interface_list[j]['currentEma'] = self.ema(a_bar, self.complete_interface_list[j]['use_averages'], self.period, self.complete_interface_list[j]['prevEma'], smoothing=None)
self.complete_interface_list[j]['prevEma'] = self.complete_interface_list[j]['currentEma']
def start_monitoring(self):
""" Starts the thread that monitors interface occupation """
self.report_object = application_switch_3.ApplicationSwitch()
self.monitoring=1
self.threads_id.append(threading.Thread(name = 'Monitor', target=self.monitor))
self.threads_id[0].start()
def stop_monitoring(self):
""" Stops monitoring the output interface """
self.monitoring=0
#toDo: Handle
def congestion_stopped(self):
""" Unused """
self.is_congested=0
def monitor(self):
""" Obtains a new sample of the interface occupation average, and in case of congestion, notifies the main module """
self.startup_time = time.time()
while True:
if self.monitoring == 1:
try:
self.update_window()
for j in range(len(self.complete_interface_list)):
#print "update, ema: " + str(self.complete_interface_list[j]['currentEma'])
#print "current threshold: " + str(self.complete_interface_list[j]['threshold'])
if (self.complete_interface_list[j]['is_congested'] == 0) and (self.complete_interface_list[j]['currentEma'] >= self.complete_interface_list[j]['threshold']):
#print "Congested"
self.detection_time = time.time()
self.complete_interface_list[j]['threshold'] = self.complete_interface_list[j]['lower_limit']
self.monitoring = 0
self.report_object.congestion_detected(self.complete_interface_list[j])
elif (self.complete_interface_list[j]['is_congested'] == 1) and (self.complete_interface_list[j]['currentEma'] <= self.complete_interface_list[j]['threshold']):
self.complete_interface_list[j]['is_congested'] = 0
self.complete_interface_list[j]['threshold'] = self.complete_interface_list[j]['upper_limit']
#print "Congestion ceased"
self.report_object.congestion_ceased()
except KeyboardInterrupt:
print " \n *** So long and thanks for all the fish! *** "
self.monitoring = 0
break
def clear_queues(self, controller_message):
subprocess.check_output('./clear_queues.sh ', shell=True)
del self.old_queue_list[:]
self.qos_register['uuid'] = None
del self.queues_ids[:]
def update_queues(self, controller_message):
""" Updates the QoS queues, one queue is created for each flow """
# Here we should:
# 0. Check if register QoS exists, if not create it
# 1. Compare the received queue list with the previous one and:
# 1a. If there are new elements, create and attach a queue for them
# 1b. If there are deleted elements, delete the queue
# 1c. If there are elements with different bw, update it
# 2. Store the old queue list as the updated one
# 3. Notify the controller about the queue completion
# Queues are at (controller_message['bw_list'])
self.lock.acquire()
to_create = []
#to_update =[]
to_delete = []
# Check if qos exists
if self.qos_register['uuid'] == None:
self.create_qos_register(self.complete_interface_list[0]['name'])
print "received bw list: ", controller_message['bw_list']
print "old bw list: ", self.old_queue_list
for j in range(len(controller_message['bw_list'])):
# Flow still exists, getting bw/s
for k in range(len(self.old_queue_list)):
if (controller_message['bw_list'][j]['nw_src'] == self.old_queue_list[k]['nw_src']) and (controller_message['bw_list'][j]['nw_dst'] == self.old_queue_list[k]['nw_dst']):
self.set_queue_bw(self.complete_interface_list[0]['name'], k, controller_message['bw_list'][j]['bw'])
break
# If it wasn't in k-1 and k we could have a) flow ceased b) flow is a new one
if (not any(src['nw_src'] == controller_message['bw_list'][j]['nw_src'] for src in self.old_queue_list)):
# New flow does not exist in the old flow stats, append it
#new_flows_indexes.append(j)
print "New queue to create: ", controller_message['bw_list'][j]
to_create.append(controller_message['bw_list'][j])
continue
if not self.old_queue_list:
print "Empty old list!"
to_create.append(controller_message['bw_list'][j])
for j in range(len(self.old_queue_list)):
if (not any(src['nw_src'] == self.old_queue_list[j]['nw_src'] for src in controller_message['bw_list'])):
# New flow does not exist in the old flow stats, append it
print "Old flows to delete: ", self.old_queue_list[j]
to_delete.append(j)
continue
self.create_individual_queues(self.complete_interface_list[0]['name'], to_create)
self.delete_individual_queue(self.complete_interface_list[0]['name'], to_delete)
self.report_object.queues_ready(self.complete_interface_list[0],controller_message['bw_list'], self.old_queue_list)
self.lock.release()
def set_queue_bw(self, interface_name, queue_index, bw):
#ovs-vsctl set Queue e059add5-ea8d-4c05-a9be-895ab217d2b4 other-config:max-rate=99
print "Giving bw of ", bw
command = 'ovs-vsctl set Queue ' + self.old_queue_list[queue_index]['uuid'] +' other-config:max-rate=' + str(bw)
subprocess.check_output(command, shell=True)
self.old_queue_list[queue_index]['bw'] = bw
def delete_a_queue(self, a_queue):
for i in range(len(self.old_queue_list)):
if (self.old_queue_list[i]['nw_src'] == a_queue['nw_src']) and (self.old_queue_list[i]['nw_dst'] == a_queue['nw_dst']):
delete_index = i
break
command = 'ovs-vsctl remove QoS ' + self.qos_register['uuid'] + ' queues ' + str(self.old_queue_list[delete_index]['queueId'])
subprocess.check_output(command, shell=True)
command = 'ovs-vsctl destroy queue ' + str(self.old_queue_list[delete_index]['uuid'])
subprocess.check_output(command, shell=True)
self.queues_ids.remove(self.old_queue_list[delete_index]['queueId'])
del self.old_queue_list[delete_index]
def delete_individual_queue(self, interface_name, to_delete):
for i in range(len(to_delete)):
command = 'ovs-vsctl list Queue ' + '| grep ' + str(self.old_queue_list[to_delete[i]]['uuid'])
result = subprocess.check_output(command, shell=True).split('\n')[0]
print "Grep command result: ", result
if not result:
continue
command = 'ovs-vsctl remove QoS ' + self.qos_register['uuid'] + ' queues ' + str(self.old_queue_list[to_delete[i]]['queueId'])
subprocess.check_output(command, shell=True)
command = 'ovs-vsctl destroy queue ' + str(self.old_queue_list[to_delete[i]]['uuid'])
subprocess.check_output(command, shell=True)
self.queues_ids.remove(self.old_queue_list[to_delete[i]]['queueId'])
removeset = set(to_delete)
newlist = [v for k, v in enumerate(self.old_queue_list) if k not in removeset]
del self.old_queue_list[:]
for j in range(len(newlist)):
self.old_queue_list.append(newlist[j])
def create_individual_queues(self, interface_name, to_create):
#queue_list = []
#print "creating queues: ", to_create
for i in range(len(to_create)):
a_queue_dict = dict.fromkeys(['uuid', 'queueId', 'nw_src', 'nw_dst', 'bw'])
a = 0
while True:
if a not in self.queues_ids:
self.queues_ids.append(a)
break
else:
a = a +1
command = 'ovs-vsctl create Queue other-config:max-rate=' + str(to_create[i]['bw'])
an_uuid = subprocess.check_output(command, shell=True).split('\n')[0]
command = 'ovs-vsctl add Qos ' + self.qos_register['uuid'] + ' queues ' + str(a) + '=' + an_uuid
subprocess.check_output(command, shell=True)
a_queue_dict['uuid'] = an_uuid
a_queue_dict['queueId'] = a
a_queue_dict['nw_src'] = to_create[i]['nw_src']
a_queue_dict['nw_dst'] = to_create[i]['nw_dst']
a_queue_dict['bw'] = to_create[i]['bw']
self.old_queue_list.append(a_queue_dict)
def create_qos_register(self, interface_name):
#ovs-vsctl -- set Port eth0br qos=@fenceqos -- --id=@fenceqos create QoS type=linux-htb
#self.qos_register = dict.fromkeys(['uuid','port', 'id', 'min-rate', 'max-rate'] )
command = 'ovs-vsctl -- set Port ' + interface_name + ' qos=@fenceqos -- --id=@fenceqos create QoS type=linux-htb'
self.qos_register['uuid'] = subprocess.check_output(command, shell=True).split('\n')[0]
self.qos_register['port'] = interface_name
self.qos_register['id'] = 'fenceqos'
self.qos_register['max-rate'] = '900000000'
#ovs-vsctl set Qos 016d2315-6305-4692-ae89-c2a3e680e874 other-config:max-rate=1000000
print "QoS uuid: ", self.qos_register['uuid']
command = 'ovs-vsctl set Qos ' + self.qos_register['uuid'] + ' other-config:max-rate=900000000'
subprocess.check_output(command, shell=True)
def create_queues(self, controller_message):
""" Creates the QoS queues, one queue is created for each flow """
self.clear_queues(controller_message)
self.queues_creation_time = time.time()
self.complete_interface_list[0]['queueList']=self.init_queues(self.complete_interface_list[0]['name'],controller_message['bw_list'])
self.set_queues_bw(self.complete_interface_list[0]['queueList'])
self.report_object.queues_ready(self.complete_interface_list[0],controller_message['bw_list'],self.complete_interface_list[0]['queueList'])
self.queues_complete_time = time.time()
#print "Startup time: " + str(self.startup_time)
#print "Detection time: " + str(self.detection_time)
#print "Queues creation time: " + str(self.queues_creation_time)
#print "Queues complete time: " + str(self.queues_complete_time)
@classmethod
def init_queues(cls, interface_name, bw_list):
""" Inits the QoS queues """
#print "Initing queues for: " + str(interface_name)
queues_list=[]
qos_string='ovs-vsctl -- set Port ' + interface_name + ' qos=@fenceqos -- --id=@fenceqos create qos type=linux-htb other-config:max-rate=900000000'
queues_string=''
for j in range(len(bw_list)):
a_queue_dict=dict.fromkeys(['queueId','queueuuid','nw_src','nw_dst','bw'])
a_queue_dict['queueId']=j
a_queue_dict['nw_src']=bw_list[j]['nw_src']
a_queue_dict['nw_dst']=bw_list[j]['nw_dst']
a_queue_dict['bw'] = bw_list[j]['bw']
a_queue= str(a_queue_dict['queueId']) +'=@queue' + str(a_queue_dict['queueId'])
if j < len(bw_list) - 1:
a_queue = a_queue + ','
queues_string=queues_string+a_queue
queues_list.append(a_queue_dict)
queues_string='queues='+ queues_string
queues_creation=''
for j in range(len(bw_list)):
a_creation='-- --id=@queue' + str(queues_list[j]['queueId']) + ' create Queue other-config:max-rate=100000000 '
queues_creation=queues_creation+a_creation
command=qos_string + ' ' + queues_string + ' ' + queues_creation
#print "Queue command: \n " + str(command)
subprocess.check_output(command, shell=True)
# Getting uuid of each queue
queues_string = subprocess.check_output("ovs-vsctl list Queue", shell=True)
#print "Queues Ready: " + str(queues_string)
allqueues_string = subprocess.check_output("ovs-vsctl list QoS | grep queues", shell=True)
for j in range(len(queues_list)):
queues_list[j]['queueuuid']=allqueues_string.split(":")[1].split(",")[j].split("=")[1].split('}\n')[0].strip()
return queues_list
@classmethod
def set_queues_bw(cls, queues_list):
""" Sets the queue bw, according to the policy defined by the SDN controller """
for i in range(len(queues_list)):
subprocess.check_output("ovs-vsctl set queue " + queues_list[i]['queueuuid'] + " other-config:max-rate="+str(queues_list[i]['bw']), shell=True)
def ema(self, a_bar, series, period, prevma, smoothing=None):
'''Returns the Exponential Moving Average of a series.
Keyword arguments:
a_bar -- currrent index or location of the series
series -- series of values to be averaged
period -- number of values in the series to average
prevma -- previous exponential moving average
smoothing -- smoothing factor to use in the series.
valid values: between 0 & 1.
default: None - which then uses formula = 2.0 / (period + 1.0)
closer to 1 to gives greater weight to recent values - less smooth
closer to 0 gives greater weight to older values -- more smooth
'''
smoothing = 0.8
if a_bar[0] <= 0:
return series[0]
elif a_bar[0] < period:
return self.cumulative_sma(a_bar[0], series, prevma)
return prevma + smoothing * (series[a_bar[0]] - prevma)
@classmethod
def cumulative_sma(cls, a_bar, series, prevma):
"""
Returns the cumulative or unweighted simple moving average.
Avoids averaging the entire series on each call.
Keyword arguments:
a_bar -- current index or location of the value in the series
series -- list or tuple of data to average
prevma -- previous average (n - 1) of the series.
"""
if a_bar[0] <= 0:
return series[0]
else:
return prevma + ((series[a_bar[0]] - prevma) / (a_bar[0] + 1.0))
def get_sample(self, interval_time=1.0):
""" Obtains a sample of the interface occupation in bytes/s """
samples_list=[]
for j in range(len(self.complete_interface_list)):
sample_dict=dict.fromkeys(['interface_name'],['sample'])
samples_list.append(sample_dict)
#lists to Store first and second sample value of each interface
# Each value of a and b represents a sample taken in each interface
sample_1 = []
sample_2 = []
for j in range(len(self.complete_interface_list)):
sample_1.append((float(subprocess.check_output("cat /proc/net/dev | grep " + self.complete_interface_list[j]['name'] + " | awk '{print $10;}'", shell=True).split('\n')[0])))
time.sleep(interval_time)
for j in range(len(self.complete_interface_list)):
sample_2.append((float(subprocess.check_output("cat /proc/net/dev | grep " + self.complete_interface_list[j]['name'] + " | awk '{print $10;}'", shell=True).split('\n')[0])))
samples_list[j]['name'] = self.complete_interface_list[j]['name']
#samples_list[j]['sample']=((b[j]-a[j])/1048576) In MBytes
samples_list[j]['sample']=sample_2[j]-sample_1[j]
return samples_list
if __name__ == "__main__":
SOME_SAMPLES = 10
PERIOD = 3 #number of bars to average
AN_INTERVAL_TIME = 1.0
#toDo: Handle this as a percentage of total link capacity
AN_UPPER_LIMIT = 0.4
LOWER_LIMIT = 0.41
USE_AVERAGES = deque( maxlen=SOME_SAMPLES )
CODE = FlowMonitor_3(SOME_SAMPLES, AN_INTERVAL_TIME, AN_UPPER_LIMIT)
CODE.start_monitoring()