-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy pathFlowMonitor_2.py
342 lines (254 loc) · 13.9 KB
/
FlowMonitor_2.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
""" 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_2:
""" 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 = []
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.reset_queues()
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)
def update_queues(self, controller_message):
""" Updates the QoS queues, one queue is created for each flow """
# Here we should:
# 1. Init and set bw with the new queue list
# 2. Update the updated queues
# 3 . Delete de deleted queues
# Create lit with new queue
# Update queue list
for i in range(len(controller_message['update_queue'])):
for j in range(len(self.complete_interface_list[0]['queueList'])):
if (controller_message['update_queue'][i]['nw_src'] == self.complete_interface_list[0]['queueList'][j]['nw_src']) and (controller_message['update_queue'][i]['nw_dst'] == self.complete_interface_list[0]['queueList'][j]['nw_dst']):
self.complete_interface_list[0]['queueList'][j]['bw'] = controller_message['update_queue'][i]['bw']
# Append new queues to the global queue list
for i in range(len(new_queue_list)):
self.complete_interface_list[0]['queueList'].append(new_queue_list[i])
delete_indexes = []
for i in range(len(controller_message['delete_queue'])):
for j in range(len(self.complete_interface_list[0]['queueList'])):
if (controller_message['delete_queue'][i]['nw_src'] == self.complete_interface_list[0]['queueList'][j]['nw_src']) and (controller_message['delete_queue'][i]['nw_dst'] == self.complete_interface_list[0]['queueList'][j]['nw_dst']):
delete_indexes.append(j)
for i in range(len(delete_indexes)):
del self.complete_interface_list[0]['queueList'][delete_indexes[i]]
new_queue_list = self.init_queues(self.complete_interface_list[0]['name'],controller_message['create_queue'])
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'])
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_2(SOME_SAMPLES, AN_INTERVAL_TIME, AN_UPPER_LIMIT)
CODE.start_monitoring()